From 80abf9d57e754bcb6b80d0501450a40bb281d3b6 Mon Sep 17 00:00:00 2001
From: Rickard Green
Date: Tue, 12 Mar 2019 22:52:31 +0100
Subject: Add possibility to also get size of data from
erlang:dist_ctrl_get_data()
---
erts/doc/src/erlang.xml | 85 ++++++++++++++++++++++++++
erts/emulator/beam/atom.names | 1 +
erts/emulator/beam/bif.tab | 2 +
erts/emulator/beam/dist.c | 114 +++++++++++++++++++++++++++++++++--
erts/emulator/beam/dist.h | 3 +
erts/emulator/beam/erl_node_tables.c | 2 +
erts/emulator/beam/erl_node_tables.h | 1 +
erts/preloaded/ebin/erlang.beam | Bin 99948 -> 100196 bytes
erts/preloaded/src/erlang.erl | 20 +++++-
9 files changed, 223 insertions(+), 5 deletions(-)
(limited to 'erts')
diff --git a/erts/doc/src/erlang.xml b/erts/doc/src/erlang.xml
index e78ded4ae1..0d94f83493 100644
--- a/erts/doc/src/erlang.xml
+++ b/erts/doc/src/erlang.xml
@@ -1269,6 +1269,42 @@ end
data is available by calling
erlang:dist_ctrl_get_data_notification(DHandle).
+ The returned value when there are data available depends
+ on the value of the get_size option configured on the
+ distribution channel identified by DHandle.
+ For more information see the documentation of the get_size
+ option for the
+ erlang:dist_ctrl_set_opt/3
+ function.
+
+ Only the process registered as distribution
+ controller for the distribution channel identified by
+ DHandle is allowed to call this
+ function.
+
+
+ This function is used when implementing an alternative
+ distribution carrier using processes as distribution
+ controllers. DHandle is retrived
+ via the callback
+ f_handshake_complete.
+ More information can be found in the documentation of
+ ERTS
+ User's Guide ➜ How to implement an Alternative Carrier
+ for the Erlang Distribution ➜ Distribution Module.
+
+
+
+
+
+
+ Get value of the get_size option on a distribution channel
+
+ Returns the value of the get_size option on the distribution channel
+ identified by DHandle. For more information see the
+ documentation of the get_size option for the
+ erlang:dist_ctrl_set_opt/3
+ function.
Only the process registered as distribution
controller for the distribution channel identified by
@@ -1391,6 +1427,55 @@ end
+
+
+ Set value of the get_size option on a distribution channel
+
+ Sets the value of the get_size option on the distribution channel
+ identified by DHandle. This option controls the return
+ value of calls to
+ erlang:dist_ctrl_get_data(DHandle)
+ where DHandle equals DHandle used
+ when setting this option.
+ When the get_size option is:
+
+ false
+ -
+ and there are distribution data available, a call to
+ erlang:dist_ctrl_get_data(DHandle)
+ will just return Data to pass over the channel.
+ This is the default value of the get_size option.
+
+ true
+ -
+ and there are distribution data available, a call to
+ erlang:dist_ctrl_get_data(DHandle)
+ will return Data to pass over the channel as well as
+ the Size of Data in bytes. This is returned as
+ a tuple on the form {Size, Data}.
+
+
+ All options are set to default when a channel is closed.
+
+ Only the process registered as distribution
+ controller for the distribution channel identified by
+ DHandle is allowed to call this
+ function.
+
+
+ This function is used when implementing an alternative
+ distribution carrier using processes as distribution
+ controllers. DHandle is retrived
+ via the callback
+ f_handshake_complete.
+ More information can be found in the documentation of
+ ERTS
+ User's Guide ➜ How to implement an Alternative Carrier
+ for the Erlang Distribution ➜ Distribution Module.
+
+
+
+
Return the Nth element of a tuple.
diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names
index f81082a698..412d689246 100644
--- a/erts/emulator/beam/atom.names
+++ b/erts/emulator/beam/atom.names
@@ -290,6 +290,7 @@ atom Ge='>='
atom generational
atom get_all_trap
atom get_seq_token
+atom get_size
atom get_tcw
atom gather_gc_info_result
atom gather_io_bytes
diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 11941db8cd..34a0be4f2d 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -160,6 +160,8 @@ bif erlang:dist_ctrl_input_handler/2
bif erlang:dist_ctrl_put_data/2
bif erlang:dist_ctrl_get_data/1
bif erlang:dist_ctrl_get_data_notification/1
+bif erlang:dist_ctrl_get_opt/2
+bif erlang:dist_ctrl_set_opt/3
# Static native functions in erts_internal
bif erts_internal:port_info/1
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index b50c8273b1..ff5f766de7 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -3252,6 +3252,86 @@ dist_ctrl_put_data_2(BIF_ALIST_2)
BIF_RET(am_ok);
}
+BIF_RETTYPE
+dist_ctrl_set_opt_3(BIF_ALIST_3)
+{
+ DistEntry *dep = ERTS_PROC_GET_DIST_ENTRY(BIF_P);
+ Uint32 conn_id;
+ BIF_RETTYPE ret;
+
+ if (!dep)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ if (erts_dhandle_to_dist_entry(BIF_ARG_1, &conn_id) != dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_de_rlock(dep);
+
+ if (dep->connection_id != conn_id)
+ ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+ else {
+
+ switch (BIF_ARG_2) {
+ case am_get_size:
+ ERTS_BIF_PREP_RET(ret, (dep->opts & ERTS_DIST_CTRL_OPT_GET_SIZE
+ ? am_true
+ : am_false));
+ if (BIF_ARG_3 == am_true)
+ dep->opts |= ERTS_DIST_CTRL_OPT_GET_SIZE;
+ else if (BIF_ARG_3 == am_false)
+ dep->opts &= ~ERTS_DIST_CTRL_OPT_GET_SIZE;
+ else
+ ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+ break;
+ default:
+ ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+ break;
+ }
+
+ }
+
+ erts_de_runlock(dep);
+
+ return ret;
+}
+
+BIF_RETTYPE
+dist_ctrl_get_opt_2(BIF_ALIST_2)
+{
+ DistEntry *dep = ERTS_PROC_GET_DIST_ENTRY(BIF_P);
+ Uint32 conn_id;
+ BIF_RETTYPE ret;
+
+ if (!dep)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ if (erts_dhandle_to_dist_entry(BIF_ARG_1, &conn_id) != dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_de_rlock(dep);
+
+ if (dep->connection_id != conn_id)
+ ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+ else {
+
+ switch (BIF_ARG_2) {
+ case am_get_size:
+ ERTS_BIF_PREP_RET(ret, (dep->opts & ERTS_DIST_CTRL_OPT_GET_SIZE
+ ? am_true
+ : am_false));
+ break;
+ default:
+ ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
+ break;
+ }
+
+ }
+
+ erts_de_runlock(dep);
+
+ return ret;
+}
+
BIF_RETTYPE
dist_get_stat_1(BIF_ALIST_1)
{
@@ -3332,7 +3412,9 @@ dist_ctrl_get_data_1(BIF_ALIST_1)
Eterm *hp;
ProcBin *pb;
erts_aint_t qsize;
- Uint32 conn_id;
+ Uint32 conn_id, get_size;
+ Eterm res;
+ Uint hsz, bin_sz;
if (!dep)
BIF_ERROR(BIF_P, EXC_NOTSUP);
@@ -3400,15 +3482,26 @@ dist_ctrl_get_data_1(BIF_ALIST_1)
erts_de_runlock(dep);
- hp = HAlloc(BIF_P, PROC_BIN_SIZE);
+ bin_sz = obuf->ext_endp - obuf->extp;
+ hsz = PROC_BIN_SIZE;
+
+ get_size = dep->opts & ERTS_DIST_CTRL_OPT_GET_SIZE;
+ if (get_size) {
+ hsz += 3; /* 2 tuple */
+ if (!IS_USMALL(0, bin_sz))
+ hsz += BIG_UINT_HEAP_SIZE;
+ }
+
+ hp = HAlloc(BIF_P, hsz);
pb = (ProcBin *) (char *) hp;
pb->thing_word = HEADER_PROC_BIN;
- pb->size = obuf->ext_endp - obuf->extp;
+ pb->size = bin_sz;
pb->next = MSO(BIF_P).first;
MSO(BIF_P).first = (struct erl_off_heap_header*) pb;
pb->val = ErtsDistOutputBuf2Binary(obuf);
pb->bytes = (byte*) obuf->extp;
pb->flags = 0;
+ hp += PROC_BIN_SIZE;
qsize = erts_atomic_add_read_nob(&dep->qsize, -size_obuf(obuf));
ASSERT(qsize >= 0);
@@ -3425,7 +3518,20 @@ dist_ctrl_get_data_1(BIF_ALIST_1)
}
}
- BIF_RET2(make_binary(pb), (initial_reds - reds));
+ res = make_binary(pb);
+
+ if (get_size) {
+ Eterm sz_term;
+ if (IS_USMALL(0, bin_sz))
+ sz_term = make_small(bin_sz);
+ else {
+ sz_term = uint_to_big(bin_sz, hp);
+ hp += BIG_UINT_HEAP_SIZE;
+ }
+ res = TUPLE2(hp, sz_term, res);
+ }
+
+ BIF_RET2(res, (initial_reds - reds));
}
void
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index c4bb967592..5bd22cc31f 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -172,6 +172,9 @@ extern int erts_is_alive;
/* Pending connection; signals can be enqueued */
#define ERTS_DSIG_PREP_PENDING 4
+/* dist_ctrl_{g,s}et_option/2 */
+#define ERTS_DIST_CTRL_OPT_GET_SIZE ((Uint32) (1 << 0))
+
#ifdef DEBUG
#define ERTS_DBG_CHK_NO_DIST_LNK(D, R, L) \
erts_dbg_chk_no_dist_proc_link((D), (R), (L))
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index afafaf48dc..1adb101e30 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -177,6 +177,7 @@ dist_table_alloc(void *dep_tmpl)
dep->connection_id = 0;
dep->state = ERTS_DE_STATE_IDLE;
dep->flags = 0;
+ dep->opts = 0;
dep->version = 0;
dep->mld = NULL;
@@ -659,6 +660,7 @@ erts_set_dist_entry_not_connected(DistEntry *dep)
dep->state = ERTS_DE_STATE_IDLE;
dep->flags = 0;
+ dep->opts = 0;
dep->prev = NULL;
dep->cid = NIL;
diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h
index d5daf0c2df..8153699596 100644
--- a/erts/emulator/beam/erl_node_tables.h
+++ b/erts/emulator/beam/erl_node_tables.h
@@ -148,6 +148,7 @@ struct dist_entry_ {
enum dist_entry_state state;
Uint32 flags; /* Distribution flags, like hidden,
atom cache etc. */
+ Uint32 opts;
unsigned long version; /* Protocol version */
ErtsMonLnkDist *mld; /* Monitors and links */
diff --git a/erts/preloaded/ebin/erlang.beam b/erts/preloaded/ebin/erlang.beam
index 1b0cb5b50c..62dc8702e7 100644
Binary files a/erts/preloaded/ebin/erlang.beam and b/erts/preloaded/ebin/erlang.beam differ
diff --git a/erts/preloaded/src/erlang.erl b/erts/preloaded/src/erlang.erl
index a5b60cc845..ac73946dc0 100644
--- a/erts/preloaded/src/erlang.erl
+++ b/erts/preloaded/src/erlang.erl
@@ -48,6 +48,8 @@
dist_ctrl_put_data/2,
dist_ctrl_get_data/1,
dist_ctrl_get_data_notification/1,
+ dist_ctrl_get_opt/2,
+ dist_ctrl_set_opt/3,
dist_get_stat/1]).
-deprecated([get_stacktrace/0,now/0]).
@@ -3326,7 +3328,8 @@ dist_ctrl_input_handler(_DHandle, _InputHandler) ->
dist_ctrl_put_data(_DHandle, _Data) ->
erlang:nif_error(undefined).
--spec erlang:dist_ctrl_get_data(DHandle) -> Data | 'none' when
+-spec erlang:dist_ctrl_get_data(DHandle) -> {Size, Data} | Data | 'none' when
+ Size :: non_neg_integer(),
DHandle :: dist_handle(),
Data :: iodata().
@@ -3339,6 +3342,21 @@ dist_ctrl_get_data(_DHandle) ->
dist_ctrl_get_data_notification(_DHandle) ->
erlang:nif_error(undefined).
+-spec erlang:dist_ctrl_set_opt(DHandle, 'get_size', Value) -> OldValue when
+ DHandle :: dist_handle(),
+ Value :: boolean(),
+ OldValue :: boolean().
+
+dist_ctrl_set_opt(_DHandle, _Opt, _Val) ->
+ erlang:nif_error(undefined).
+
+-spec erlang:dist_ctrl_get_opt(DHandle, 'get_size') -> Value when
+ DHandle :: dist_handle(),
+ Value :: boolean().
+
+dist_ctrl_get_opt(_DHandle, _Opt) ->
+ erlang:nif_error(undefined).
+
-spec erlang:dist_get_stat(DHandle) -> Res when
DHandle :: dist_handle(),
InputPackets :: non_neg_integer(),
--
cgit v1.2.3