aboutsummaryrefslogtreecommitdiffstats
path: root/erts
diff options
context:
space:
mode:
Diffstat (limited to 'erts')
-rw-r--r--erts/emulator/beam/erl_drv_nif.h3
-rw-r--r--erts/emulator/beam/erl_message.c4
-rw-r--r--erts/emulator/sys/common/erl_check_io.c183
-rw-r--r--erts/emulator/sys/common/erl_check_io.h3
-rw-r--r--erts/emulator/test/nif_SUITE.erl70
5 files changed, 146 insertions, 117 deletions
diff --git a/erts/emulator/beam/erl_drv_nif.h b/erts/emulator/beam/erl_drv_nif.h
index 9ef7c39d41..a5ecbfff06 100644
--- a/erts/emulator/beam/erl_drv_nif.h
+++ b/erts/emulator/beam/erl_drv_nif.h
@@ -54,7 +54,8 @@ enum ErlNifSelectFlags {
ERL_NIF_SELECT_READ = (1 << 0),
ERL_NIF_SELECT_WRITE = (1 << 1),
ERL_NIF_SELECT_STOP = (1 << 2),
- ERL_NIF_SELECT_CANCEL = (1 << 3)
+ ERL_NIF_SELECT_CANCEL = (1 << 3),
+ ERL_NIF_SELECT_CUSTOM_MSG= (1 << 4)
};
/*
diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c
index a3274d7443..942bec84cf 100644
--- a/erts/emulator/beam/erl_message.c
+++ b/erts/emulator/beam/erl_message.c
@@ -532,9 +532,7 @@ erts_try_alloc_message_on_heap(Process *pp,
if ((*psp) & ERTS_PSFLGS_VOLATILE_HEAP)
goto in_message_fragment;
- else if (
- *plp & ERTS_PROC_LOCK_MAIN
- ) {
+ else if (*plp & ERTS_PROC_LOCK_MAIN) {
try_on_heap:
if (((*psp) & ERTS_PSFLGS_VOLATILE_HEAP)
|| (pp->flags & F_DISABLE_GC)
diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c
index b4609007c9..487b9cba40 100644
--- a/erts/emulator/sys/common/erl_check_io.c
+++ b/erts/emulator/sys/common/erl_check_io.c
@@ -591,6 +591,90 @@ abort_tasks(ErtsDrvEventState *state, int mode)
}
}
+static void prepare_select_msg(struct erts_nif_select_event* e,
+ enum ErlNifSelectFlags mode,
+ Eterm recipient,
+ ErtsResource* resource,
+ Eterm msg,
+ Eterm event_atom)
+{
+ ErtsMessage* mp;
+ Eterm* hp;
+ Eterm* hp_start;
+ Uint hsz;
+
+ if (is_not_nil(e->pid)) {
+ ASSERT(e->mp);
+ erts_cleanup_messages(e->mp);
+ }
+
+ if (mode & ERL_NIF_SELECT_CUSTOM_MSG) {
+ hsz = size_object(msg);
+ mp = erts_alloc_message(hsz, &hp);
+ hp_start = hp;
+ ERL_MESSAGE_TERM(mp) = copy_struct(msg, hsz, &hp, &mp->hfrag.off_heap);
+ }
+ else {
+ ErtsBinary* bin;
+ Eterm resource_term, ref_term, tuple;
+
+ /* {select, Resource, Ref, EventAtom} */
+ hsz = 5 + ERTS_MAGIC_REF_THING_SIZE;
+ if (is_internal_ref(msg))
+ hsz += ERTS_REF_THING_SIZE;
+ else
+ ASSERT(is_immed(msg));
+
+ mp = erts_alloc_message(hsz, &hp);
+ hp_start = hp;
+
+ bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource);
+ resource_term = erts_mk_magic_ref(&hp, &mp->hfrag.off_heap, &bin->binary);
+ if (is_internal_ref(msg)) {
+ Uint32* refn = internal_ref_numbers(msg);
+ write_ref_thing(hp, refn[0], refn[1], refn[2]);
+ ref_term = make_internal_ref(hp);
+ hp += ERTS_REF_THING_SIZE;
+ }
+ else {
+ ASSERT(is_immed(msg));
+ ref_term = msg;
+ }
+ tuple = TUPLE4(hp, am_select, resource_term, ref_term, event_atom);
+ hp += 5;
+ ERL_MESSAGE_TERM(mp) = tuple;
+ }
+ ASSERT(hp == hp_start + hsz); (void)hp_start;
+
+ ASSERT(is_not_nil(recipient));
+ e->pid = recipient;
+ e->mp = mp;
+}
+
+static ERTS_INLINE void send_select_msg(struct erts_nif_select_event* e)
+{
+ Process* rp = erts_proc_lookup(e->pid);
+
+ ASSERT(is_internal_pid(e->pid));
+ if (!rp) {
+ erts_cleanup_messages(e->mp);
+ return;
+ }
+
+ erts_queue_message(rp, 0, e->mp, ERL_MESSAGE_TERM(e->mp), am_system);
+}
+
+static void clear_select_event(struct erts_nif_select_event* e)
+{
+ if (is_not_nil(e->pid)) {
+ /* Discard unsent message */
+ ASSERT(e->mp);
+ erts_cleanup_messages(e->mp);
+ e->mp = NULL;
+ e->pid = NIL;
+ }
+}
+
static void
deselect(ErtsDrvEventState *state, int mode)
{
@@ -621,8 +705,8 @@ deselect(ErtsDrvEventState *state, int mode)
erts_io_control(state, ERTS_POLL_OP_DEL, 0);
switch (state->type) {
case ERTS_EV_TYPE_NIF:
- state->driver.nif->in.pid = NIL;
- state->driver.nif->out.pid = NIL;
+ clear_select_event(&state->driver.nif->in);
+ clear_select_event(&state->driver.nif->out);
enif_release_resource(state->driver.stop.resource->data);
state->driver.stop.resource = NULL;
break;
@@ -948,7 +1032,7 @@ enif_select(ErlNifEnv* env,
enum ErlNifSelectFlags mode,
void* obj,
const ErlNifPid* pid,
- Eterm ref)
+ Eterm msg)
{
int on;
ErtsResource* resource = DATA_TO_RESOURCE(obj);
@@ -966,7 +1050,7 @@ enif_select(ErlNifEnv* env,
#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
if (!grow_drv_ev_state(fd)) {
- if (fd > 0) nif_select_large_fd_error(fd, mode, resource, ref);
+ if (fd > 0) nif_select_large_fd_error(fd, mode, resource, msg);
return INT_MIN | ERL_NIF_SELECT_INVALID_EVENT;
}
#endif
@@ -1012,21 +1096,21 @@ enif_select(ErlNifEnv* env,
* Changing process and/or ref is ok (I think?).
*/
if (state->driver.stop.resource != resource)
- nif_select_steal(state, ERL_DRV_READ | ERL_DRV_WRITE, resource, ref);
+ nif_select_steal(state, ERL_DRV_READ | ERL_DRV_WRITE, resource, msg);
break;
case ERTS_EV_TYPE_DRV_SEL:
- nif_select_steal(state, mode, resource, ref);
+ nif_select_steal(state, mode, resource, msg);
break;
case ERTS_EV_TYPE_STOP_USE: {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_nif_select_op(dsbufp, fd, mode, resource, ref);
+ print_nif_select_op(dsbufp, fd, mode, resource, msg);
steal_pending_stop_use(dsbufp, ERTS_INVALID_ERL_DRV_PORT, state, mode, on);
ASSERT(state->type == ERTS_EV_TYPE_NONE);
break;
}
case ERTS_EV_TYPE_STOP_NIF: {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_nif_select_op(dsbufp, fd, mode, resource, ref);
+ print_nif_select_op(dsbufp, fd, mode, resource, msg);
steal_pending_stop_nif(dsbufp, resource, state, mode, on);
if (state->type == ERTS_EV_TYPE_STOP_NIF) {
ret = ERL_NIF_SELECT_STOP_SCHEDULED; /* ?? */
@@ -1082,7 +1166,6 @@ enif_select(ErlNifEnv* env,
if (on) {
const Eterm recipient = pid ? pid->pid : env->proc->common.id;
- Uint32* refn;
if (!state->driver.nif)
state->driver.nif = alloc_nif_select_data();
if (state->type == ERTS_EV_TYPE_NONE) {
@@ -1093,28 +1176,12 @@ enif_select(ErlNifEnv* env,
ASSERT(state->type == ERTS_EV_TYPE_NIF);
ASSERT(state->driver.stop.resource == resource);
if (mode & ERL_DRV_READ) {
- state->driver.nif->in.pid = recipient;
- if (is_immed(ref)) {
- state->driver.nif->in.immed = ref;
- } else {
- ASSERT(is_internal_ref(ref));
- refn = internal_ref_numbers(ref);
- state->driver.nif->in.immed = THE_NON_VALUE;
- sys_memcpy(state->driver.nif->in.refn, refn,
- sizeof(state->driver.nif->in.refn));
- }
+ prepare_select_msg(&state->driver.nif->in, mode, recipient,
+ resource, msg, am_ready_input);
}
if (mode & ERL_DRV_WRITE) {
- state->driver.nif->out.pid = recipient;
- if (is_immed(ref)) {
- state->driver.nif->out.immed = ref;
- } else {
- ASSERT(is_internal_ref(ref));
- refn = internal_ref_numbers(ref);
- state->driver.nif->out.immed = THE_NON_VALUE;
- sys_memcpy(state->driver.nif->out.refn, refn,
- sizeof(state->driver.nif->out.refn));
- }
+ prepare_select_msg(&state->driver.nif->out, mode, recipient,
+ resource, msg, am_ready_output);
}
ret = 0;
}
@@ -1123,12 +1190,12 @@ enif_select(ErlNifEnv* env,
if (state->type == ERTS_EV_TYPE_NIF) {
if (mode & ERL_NIF_SELECT_READ
&& is_not_nil(state->driver.nif->in.pid)) {
- state->driver.nif->in.pid = NIL;
+ clear_select_event(&state->driver.nif->in);
ret |= ERL_NIF_SELECT_READ_CANCELLED;
}
if (mode & ERL_NIF_SELECT_WRITE
&& is_not_nil(state->driver.nif->out.pid)) {
- state->driver.nif->out.pid = NIL;
+ clear_select_event(&state->driver.nif->out);
ret |= ERL_NIF_SELECT_WRITE_CANCELLED;
}
}
@@ -1545,53 +1612,6 @@ oready(Eterm id, ErtsDrvEventState *state)
}
}
-static ERTS_INLINE void
-send_event_tuple(struct erts_nif_select_event* e, ErtsResource* resource,
- Eterm event_atom)
-{
- Process* rp = erts_proc_lookup(e->pid);
- ErtsProcLocks rp_locks = 0;
- ErtsMessage* mp;
- ErlOffHeap* ohp;
- ErtsBinary* bin;
- Eterm* hp;
- Uint hsz;
- Eterm resource_term, ref_term, tuple;
-
- if (!rp) {
- return;
- }
-
- bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource);
-
- /* {select, Resource, Ref, EventAtom} */
- if (is_value(e->immed)) {
- hsz = 5 + ERTS_MAGIC_REF_THING_SIZE;
- }
- else {
- hsz = 5 + ERTS_MAGIC_REF_THING_SIZE + ERTS_REF_THING_SIZE;
- }
-
- mp = erts_alloc_message_heap(rp, &rp_locks, hsz, &hp, &ohp);
-
- resource_term = erts_mk_magic_ref(&hp, ohp, &bin->binary);
- if (is_value(e->immed)) {
- ASSERT(is_immed(e->immed));
- ref_term = e->immed;
- }
- else {
- write_ref_thing(hp, e->refn[0], e->refn[1], e->refn[2]);
- ref_term = make_internal_ref(hp);
- hp += ERTS_REF_THING_SIZE;
- }
- tuple = TUPLE4(hp, am_select, resource_term, ref_term, event_atom);
-
- erts_queue_message(rp, rp_locks, mp, tuple, am_system);
-
- if (rp_locks)
- erts_proc_unlock(rp, rp_locks);
-}
-
static void bad_fd_in_pollset(ErtsDrvEventState *, Eterm inport, Eterm outport);
void
@@ -1765,7 +1785,6 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time)
case ERTS_EV_TYPE_NIF: { /* Requested via enif_select()... */
struct erts_nif_select_event in = {NIL};
struct erts_nif_select_event out = {NIL};
- ErtsResource* resource = NULL;
if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) {
if (revents & ERTS_POLL_EV_OUT) {
@@ -1773,6 +1792,7 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time)
out = state->driver.nif->out;
resource = state->driver.stop.resource;
state->driver.nif->out.pid = NIL;
+ state->driver.nif->out.mp = NULL;
}
}
if (revents & ERTS_POLL_EV_IN) {
@@ -1780,6 +1800,7 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time)
in = state->driver.nif->in;
resource = state->driver.stop.resource;
state->driver.nif->in.pid = NIL;
+ state->driver.nif->in.mp = NULL;
}
}
state->events &= ~revents;
@@ -1792,10 +1813,10 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time)
erts_mtx_unlock(fd_mtx(fd));
if (is_not_nil(in.pid)) {
- send_event_tuple(&in, resource, am_ready_input);
+ send_select_msg(&in);
}
if (is_not_nil(out.pid)) {
- send_event_tuple(&out, resource, am_ready_output);
+ send_select_msg(&out);
}
continue;
}
diff --git a/erts/emulator/sys/common/erl_check_io.h b/erts/emulator/sys/common/erl_check_io.h
index 31182be5ec..0f3fc4f7a2 100644
--- a/erts/emulator/sys/common/erl_check_io.h
+++ b/erts/emulator/sys/common/erl_check_io.h
@@ -138,8 +138,7 @@ typedef struct {
struct erts_nif_select_event {
Eterm pid;
- Eterm immed;
- Uint32 refn[ERTS_REF_NUMBERS];
+ ErtsMessage *mp;
};
typedef struct {
diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index edad62a9fb..8de6d256c5 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -486,6 +486,7 @@ t_on_load(Config) when is_list(Config) ->
-define(ERL_NIF_SELECT_WRITE, (1 bsl 1)).
-define(ERL_NIF_SELECT_STOP, (1 bsl 2)).
-define(ERL_NIF_SELECT_CANCEL, (1 bsl 3)).
+-define(ERL_NIF_SELECT_CUSTOM_MSG, (1 bsl 4)).
-define(ERL_NIF_SELECT_STOP_CALLED, (1 bsl 0)).
-define(ERL_NIF_SELECT_STOP_SCHEDULED, (1 bsl 1)).
@@ -496,33 +497,38 @@ t_on_load(Config) when is_list(Config) ->
select(Config) when is_list(Config) ->
ensure_lib_loaded(Config),
+ select_do(0, make_ref(), make_ref()),
- Ref = make_ref(),
- Ref2 = make_ref(),
+ RefBin = list_to_binary(lists:duplicate(100, $x)),
+ select_do(?ERL_NIF_SELECT_CUSTOM_MSG,
+ small, {a, tuple, with, "some", RefBin}),
+ ok.
+
+select_do(Flag, Ref, Ref2) ->
{{R, R_ptr}, {W, W_ptr}} = pipe_nif(),
ok = write_nif(W, <<"hej">>),
<<"hej">> = read_nif(R, 3),
%% Wait for read
eagain = read_nif(R, 3),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ bor Flag, R,null,Ref),
[] = flush(0),
ok = write_nif(W, <<"hej">>),
- [{select, R, Ref, ready_input}] = flush(),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref2),
- [{select, R, Ref2, ready_input}] = flush(),
+ receive_ready(R, Ref, ready_input),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ bor Flag,R,self(),Ref2),
+ receive_ready(R, Ref2, ready_input),
Papa = self(),
Pid = spawn_link(fun() ->
- [{select, R, Ref, ready_input}] = flush(),
+ receive_ready(R, Ref, ready_input),
Papa ! {self(), done}
end),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Pid,Ref),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, Pid, Ref),
{Pid, done} = receive_any(1000),
%% Cancel read
0 = select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref),
<<"hej">> = read_nif(R, 3),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref),
?ERL_NIF_SELECT_READ_CANCELLED =
select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref),
ok = write_nif(W, <<"hej again">>),
@@ -531,65 +537,63 @@ select(Config) when is_list(Config) ->
%% Wait for write
Written = write_full(W, $a),
- 0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,self(),Ref),
+ 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor Flag, W, self(), Ref),
[] = flush(0),
Written = read_nif(R,byte_size(Written)),
- [{select, W, Ref, ready_output}] = flush(),
+ receive_ready(W, Ref, ready_output),
%% Cancel write
- 0 = select_nif(W,?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL,W,null,Ref),
+ 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL, W, null, Ref),
Written2 = write_full(W, $b),
- 0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,null,Ref),
+ 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor Flag, W, null, Ref),
?ERL_NIF_SELECT_WRITE_CANCELLED =
- select_nif(W,?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL,W,null,Ref),
+ select_nif(W, ?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL, W, null, Ref),
Written2 = read_nif(R,byte_size(Written2)),
[] = flush(0),
%% Close write and wait for EOF
eagain = read_nif(R, 1),
- check_stop_ret(select_nif(W,?ERL_NIF_SELECT_STOP,W,null,Ref)),
+ check_stop_ret(select_nif(W, ?ERL_NIF_SELECT_STOP, W, null, Ref)),
[{fd_resource_stop, W_ptr, _}] = flush(),
{1, {W_ptr,_}} = last_fd_stop_call(),
true = is_closed_nif(W),
[] = flush(0),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref),
- [{select, R, Ref, ready_input}] = flush(),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, self(), Ref),
+ receive_ready(R, Ref, ready_input),
eof = read_nif(R,1),
- check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,null,Ref)),
+ check_stop_ret(select_nif(R, ?ERL_NIF_SELECT_STOP, R, null, Ref)),
[{fd_resource_stop, R_ptr, _}] = flush(),
{1, {R_ptr,_}} = last_fd_stop_call(),
true = is_closed_nif(R),
- select_2(Config).
+ select_2(Flag, Ref, Ref2).
-select_2(Config) ->
+select_2(Flag, Ref1, Ref2) ->
erlang:garbage_collect(),
{_,_,2} = last_resource_dtor_call(),
- Ref1 = make_ref(),
- Ref2 = make_ref(),
{{R, R_ptr}, {W, W_ptr}} = pipe_nif(),
%% Change ref
eagain = read_nif(R, 1),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref2),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, self(), Ref2),
[] = flush(0),
ok = write_nif(W, <<"hej">>),
- [{select, R, Ref2, ready_input}] = flush(),
+ receive_ready(R, Ref2, ready_input),
<<"hej">> = read_nif(R, 3),
%% Change pid
eagain = read_nif(R, 1),
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1),
Papa = self(),
spawn_link(fun() ->
- 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
+ 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1),
[] = flush(0),
Papa ! sync,
- [{select, R, Ref1, ready_input}] = flush(),
+ receive_ready(R, Ref1, ready_input),
<<"hej">> = read_nif(R, 3),
Papa ! done
end),
@@ -609,13 +613,19 @@ select_2(Config) ->
{1, {W_ptr,1}} = last_fd_stop_call(),
true = is_closed_nif(W),
- select_3(Config).
+ select_3().
-select_3(_Config) ->
+select_3() ->
erlang:garbage_collect(),
{_,_,2} = last_resource_dtor_call(),
ok.
+receive_ready(R, Ref, IOatom) when is_reference(Ref) ->
+ [{select, R, Ref, IOatom}] = flush();
+receive_ready(_, Msg, _) ->
+ [Got] = flush(),
+ {true,_,_} = {Got=:=Msg, Got, Msg}.
+
%% @doc The stealing child process for the select_steal test. Duplicates given
%% W/RFds and runs select on them to steal
select_steal_child_process(Parent, RFd) ->
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-05 03:14:17 +0000