From 75fdce43ef567668bb89508b9b8ce0df7efaa569 Mon Sep 17 00:00:00 2001 From: Sverker Eriksson Date: Mon, 6 Feb 2017 17:15:52 +0100 Subject: erts: Add enif_monitor_process and enif_demonitor_process --- erts/doc/src/erl_nif.xml | 67 +++- erts/emulator/beam/bif.c | 4 +- erts/emulator/beam/break.c | 24 +- erts/emulator/beam/dist.c | 25 +- erts/emulator/beam/erl_bif_info.c | 90 +++-- erts/emulator/beam/erl_bits.c | 3 - erts/emulator/beam/erl_driver.h | 7 - erts/emulator/beam/erl_drv_nif.h | 8 + erts/emulator/beam/erl_lock_check.c | 3 + erts/emulator/beam/erl_monitors.c | 47 ++- erts/emulator/beam/erl_monitors.h | 19 +- erts/emulator/beam/erl_nif.c | 402 ++++++++++++++++++- erts/emulator/beam/erl_nif.h | 22 +- erts/emulator/beam/erl_nif_api_funcs.h | 4 + erts/emulator/beam/erl_node_tables.c | 4 +- erts/emulator/beam/erl_process.c | 50 +-- erts/emulator/beam/erl_time_sup.c | 2 +- erts/emulator/beam/global.h | 34 +- erts/emulator/beam/io.c | 20 +- erts/emulator/sys/common/erl_check_io.c | 2 + erts/emulator/test/nif_SUITE.erl | 248 +++++++++++- erts/emulator/test/nif_SUITE_data/nif_SUITE.c | 540 +++++++++++++++++++++++++- erts/emulator/test/nif_SUITE_data/nif_mod.c | 2 + 23 files changed, 1454 insertions(+), 173 deletions(-) diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml index 8cdfc80879..bd72bdb691 100644 --- a/erts/doc/src/erl_nif.xml +++ b/erts/doc/src/erl_nif.xml @@ -711,6 +711,21 @@ typedef struct { typedef void ErlNifResourceDtor(ErlNifEnv* env, void* obj);

The function prototype of a resource destructor function.

+

The obj argument is a pointer to the resource. The only + allowed use for the resource in the destructor is to access its + user data one final time. The destructor is guaranteed to be the + last callback before the resource is deallocated.

+ + ErlNifResourceDown + + +typedef void ErlNifResourceDown(ErlNifEnv* env, void* obj, const ErlNifPid* pid, const ErlNifMonitor* mon); +

The function prototype of a resource down function, + called on the behalf of + enif_monitor_process. obj is the resource, pid + is the identity of the monitored process that is exiting, and mon + is the identity of the monitor. +

ErlNifResourceStop @@ -1021,6 +1036,29 @@ typedef enum { + + intenif_demonitor_process(ErlNifEnv* env, void* obj, + const ErlNifMonitor* mon) + Cancel a process monitor. + + +

Cancels a monitor created earlier with + enif_monitor_process. Argument obj is a pointer + to the resource holding the monitor and *mon identifies the monitor.

+

Returns 0 if the monitor was successfully identified and removed. + Returns a non-zero value if the monitor could not be identified, which means + it was either

+ + never created for this resource + already triggered + just about to be triggered by a concurrent thread + +

This function is only thread-safe when the emulator with SMP support + is used. It can only be used in a non-SMP emulator from a NIF-calling + thread.

+ + + int enif_equal_tids(ErlNifTid tid1, ErlNifTid tid2) @@ -2136,6 +2174,31 @@ enif_map_iterator_destroy(env, &iter); + + intenif_monitor_process(ErlNifEnv* env, void* obj, + const ErlNifPid* target_pid, ErlNifMonitor* mon) + Monitor a process from a resource. + + +

Starts monitoring a process from a resource. When a process is + monitored, a process exit results in a call to the provided + + down callback associated with the resource type.

+

Argument obj is pointer to the resource to hold the monitor and + *target_pid identifies the local process to be monitored.

+

If mon is not NULL, a successful call stores the + identity of the monitor in the struct pointed to by mon. + This identifier is used to refer to the monitor for later removal or + compare. A monitor is automatically removed when it triggers or when + the resource is deallocated.

+

Returns 0 on success, < 0 if no down callback is + provided, and > 0 if the process is no longer alive.

+

This function is only thread-safe when the emulator with SMP support + is used. It can only be used in a non-SMP emulator from a NIF-calling + thread.

+ + + ErlNifTime enif_monotonic_time(ErlNifTimeUnit time_unit) @@ -2268,8 +2331,8 @@ enif_map_iterator_destroy(env, &iter); used together with enif_select.

Argument init is a pointer to an ErlNifResourceTypeInit - structure that contains the function pointers for the destructor and the stop callback - of the resource type.

+ structure that contains the function pointers for destructor, down and stop callbacks + for the resource type.

 diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c index 65c370c55b..51057c3ebb 100644 --- a/erts/emulator/beam/bif.c +++ b/erts/emulator/beam/bif.c @@ -361,7 +361,7 @@ remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to) c_p->common.id, (mon->name != NIL ? mon->name - : mon->pid), + : mon->u.pid), ref, 0); res = (code == ERTS_DSIG_SEND_YIELD ? am_yield : am_true); @@ -498,7 +498,7 @@ BIF_RETTYPE demonitor(Process *c_p, Eterm ref, Eterm *multip) res = am_true; break; case MON_ORIGIN: - to = mon->pid; + to = mon->u.pid; *multip = am_false; if (is_atom(to)) { /* Monitoring a name at node to */ diff --git a/erts/emulator/beam/break.c b/erts/emulator/beam/break.c index 6e1e94b95b..5509c50a8e 100644 --- a/erts/emulator/beam/break.c +++ b/erts/emulator/beam/break.c @@ -178,19 +178,29 @@ static void doit_print_monitor(ErtsMonitor *mon, void *vpcontext) prefix = ""; } - if (mon->type == MON_ORIGIN) { - if (is_atom(mon->pid)) { /* dist by name */ - ASSERT(is_node_name_atom(mon->pid)); + switch (mon->type) { + case MON_ORIGIN: + if (is_atom(mon->u.pid)) { /* dist by name */ + ASSERT(is_node_name_atom(mon->u.pid)); erts_print(to, to_arg, "%s{to,{%T,%T},%T}", prefix, mon->name, - mon->pid, mon->ref); + mon->u.pid, mon->ref); } else if (is_atom(mon->name)){ /* local by name */ erts_print(to, to_arg, "%s{to,{%T,%T},%T}", prefix, mon->name, erts_this_dist_entry->sysname, mon->ref); } else { /* local and distributed by pid */ - erts_print(to, to_arg, "%s{to,%T,%T}", prefix, mon->pid, mon->ref); + erts_print(to, to_arg, "%s{to,%T,%T}", prefix, mon->u.pid, mon->ref); } - } else { /* MON_TARGET */ - erts_print(to, to_arg, "%s{from,%T,%T}", prefix, mon->pid, mon->ref); + break; + case MON_TARGET: + erts_print(to, to_arg, "%s{from,%T,%T}", prefix, mon->u.pid, mon->ref); + break; + case MON_NIF_TARGET: { + ErtsResource* rsrc = mon->u.resource; + /*SVERK: Print resource-ref? */ + erts_print(to, to_arg, "%s{from,{%T,%T},%T}", prefix, rsrc->type->module, + rsrc->type->name, mon->ref); + break; + } } } diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c index 9e8f853279..016abaf717 100644 --- a/erts/emulator/beam/dist.c +++ b/erts/emulator/beam/dist.c @@ -259,7 +259,7 @@ static void doit_monitor_net_exits(ErtsMonitor *mon, void *vnecp) DistEntry *dep = ((NetExitsContext *) vnecp)->dep; ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK; - rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks); + rp = erts_pid2proc(NULL, 0, mon->u.pid, rp_locks); if (!rp) goto done; @@ -278,10 +278,11 @@ static void doit_monitor_net_exits(ErtsMonitor *mon, void *vnecp) rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref); /* ASSERT(rmon != NULL); can happen during process exit */ if (rmon != NULL) { + ASSERT(rmon->type == MON_ORIGIN); ASSERT(is_atom(rmon->name) || is_nil(rmon->name)); watched = (is_atom(rmon->name) ? TUPLE2(lhp, rmon->name, dep->sysname) - : rmon->pid); + : rmon->u.pid); #ifdef ERTS_SMP rp_locks |= ERTS_PROC_LOCKS_MSG_SEND; erts_smp_proc_lock(rp, ERTS_PROC_LOCKS_MSG_SEND); @@ -1391,7 +1392,7 @@ int erts_net_message(Port *prt, if (mon == NULL) { break; } - watched = mon->pid; + watched = mon->u.pid; erts_destroy_monitor(mon); rp = erts_pid2proc_opt(NULL, 0, watched, ERTS_PROC_LOCK_LINK, @@ -1549,7 +1550,7 @@ int erts_net_message(Port *prt, if (mon == NULL) { break; } - rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks); + rp = erts_pid2proc(NULL, 0, mon->u.pid, rp_locks); erts_destroy_monitor(mon); if (rp == NULL) { @@ -1566,7 +1567,7 @@ int erts_net_message(Port *prt, watched = (is_not_nil(mon->name) ? TUPLE2(&lhp[0], mon->name, sysname) - : mon->pid); + : mon->u.pid); erts_queue_monitor_message(rp, &rp_locks, ref, am_process, watched, reason); @@ -2415,21 +2416,21 @@ static void doit_print_monitor_info(ErtsMonitor *mon, void *vptdp) void *arg = ((struct print_to_data *) vptdp)->arg; Process *rp; ErtsMonitor *rmon; - rp = erts_proc_lookup(mon->pid); + rp = erts_proc_lookup(mon->u.pid); if (!rp || (rmon = erts_lookup_monitor(ERTS_P_MONITORS(rp), mon->ref)) == NULL) { - erts_print(to, arg, "Warning, stray monitor for: %T\n", mon->pid); + erts_print(to, arg, "Warning, stray monitor for: %T\n", mon->u.pid); } else if (mon->type == MON_ORIGIN) { /* Local pid is being monitored */ erts_print(to, arg, "Remotely monitored by: %T %T\n", - mon->pid, rmon->pid); + mon->u.pid, rmon->u.pid); } else { - erts_print(to, arg, "Remote monitoring: %T ", mon->pid); - if (is_not_atom(rmon->pid)) - erts_print(to, arg, "%T\n", rmon->pid); + erts_print(to, arg, "Remote monitoring: %T ", mon->u.pid); + if (is_not_atom(rmon->u.pid)) + erts_print(to, arg, "%T\n", rmon->u.pid); else erts_print(to, arg, "{%T, %T}\n", rmon->name, - rmon->pid); /* which in this case is the + rmon->u.pid); /* which in this case is the remote system name... */ } } diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c index 5ee19aead8..9f32e0e7bd 100644 --- a/erts/emulator/beam/erl_bif_info.c +++ b/erts/emulator/beam/erl_bif_info.c @@ -202,8 +202,10 @@ bld_bin_list(Uint **hpp, Uint *szp, ErlOffHeap* oh) static void do_calc_mon_size(ErtsMonitor *mon, void *vpsz) { Uint *psz = vpsz; - *psz += IS_CONST(mon->ref) ? 0 : NC_HEAP_SIZE(mon->ref); - *psz += IS_CONST(mon->pid) ? 0 : NC_HEAP_SIZE(mon->pid); + *psz += NC_HEAP_SIZE(mon->ref); + *psz += (mon->type == MON_NIF_TARGET ? + erts_resource_ref_size(mon->u.resource) : + (is_immed(mon->u.pid) ? 0 : NC_HEAP_SIZE(mon->u.pid))); *psz += 8; /* CONS + 5-tuple */ } @@ -218,12 +220,11 @@ static void do_make_one_mon_element(ErtsMonitor *mon, void * vpmlc) { MonListContext *pmlc = vpmlc; Eterm tup; - Eterm r = (IS_CONST(mon->ref) - ? mon->ref - : STORE_NC(&(pmlc->hp), &MSO(pmlc->p), mon->ref)); - Eterm p = (IS_CONST(mon->pid) - ? mon->pid - : STORE_NC(&(pmlc->hp), &MSO(pmlc->p), mon->pid)); + Eterm r = STORE_NC(&(pmlc->hp), &MSO(pmlc->p), mon->ref); + Eterm p = (mon->type == MON_NIF_TARGET ? + erts_bld_resource_ref(&(pmlc->hp), &MSO(pmlc->p), mon->u.resource) + : (is_immed(mon->u.pid) ? mon->u.pid + : STORE_NC(&(pmlc->hp), &MSO(pmlc->p), mon->u.pid))); tup = TUPLE5(pmlc->hp, pmlc->tag, make_small(mon->type), r, p, mon->name); pmlc->hp += 6; pmlc->res = CONS(pmlc->hp, tup, pmlc->res); @@ -262,7 +263,7 @@ make_monitor_list(Process *p, ErtsMonitor *root) static void do_calc_lnk_size(ErtsLink *lnk, void *vpsz) { Uint *psz = vpsz; - *psz += IS_CONST(lnk->pid) ? 0 : NC_HEAP_SIZE(lnk->pid); + *psz += is_immed(lnk->pid) ? 0 : NC_HEAP_SIZE(lnk->pid); if (lnk->type != LINK_NODE && ERTS_LINK_ROOT(lnk) != NULL) { /* Node links use this pointer as ref counter... */ erts_doforall_links(ERTS_LINK_ROOT(lnk),&do_calc_lnk_size,vpsz); @@ -282,7 +283,7 @@ static void do_make_one_lnk_element(ErtsLink *lnk, void * vpllc) LnkListContext *pllc = vpllc; Eterm tup; Eterm old_res, targets = NIL; - Eterm p = (IS_CONST(lnk->pid) + Eterm p = (is_immed(lnk->pid) ? lnk->pid : STORE_NC(&(pllc->hp), &MSO(pllc->p), lnk->pid)); if (lnk->type == LINK_NODE) { @@ -366,8 +367,12 @@ erts_print_system_version(fmtfn_t to, void *arg, Process *c_p) typedef struct { /* {Entity,Node} = {monitor.Name,monitor.Pid} for external by name * {Entity,Node} = {monitor.Pid,NIL} for external/external by pid - * {Entity,Node} = {monitor.Name,erlang:node()} for internal by name */ - Eterm entity; + * {Entity,Node} = {monitor.Name,erlang:node()} for internal by name + * {Entity,Node} = {monitor.resource,MON_NIF_TARGET}*/ + union { + Eterm term; + ErtsResource* resource; + }entity; Eterm node; /* pid is actual target being monitored, no matter pid/port or name */ Eterm pid; @@ -413,7 +418,7 @@ static void collect_one_link(ErtsLink *lnk, void *vmicp) if (!(lnk->type == LINK_PID)) { return; } - micp->mi[micp->mi_i].entity = lnk->pid; + micp->mi[micp->mi_i].entity.term = lnk->pid; micp->sz += 2 + NC_HEAP_SIZE(lnk->pid); micp->mi_i++; } @@ -426,20 +431,20 @@ static void collect_one_origin_monitor(ErtsMonitor *mon, void *vmicp) return; } EXTEND_MONITOR_INFOS(micp); - if (is_atom(mon->pid)) { /* external by name */ - micp->mi[micp->mi_i].entity = mon->name; - micp->mi[micp->mi_i].node = mon->pid; + if (is_atom(mon->u.pid)) { /* external by name */ + micp->mi[micp->mi_i].entity.term = mon->name; + micp->mi[micp->mi_i].node = mon->u.pid; micp->sz += 3; /* need one 2-tuple */ - } else if (is_external_pid(mon->pid)) { /* external by pid */ - micp->mi[micp->mi_i].entity = mon->pid; + } else if (is_external_pid(mon->u.pid)) { /* external by pid */ + micp->mi[micp->mi_i].entity.term = mon->u.pid; micp->mi[micp->mi_i].node = NIL; - micp->sz += NC_HEAP_SIZE(mon->pid); + micp->sz += NC_HEAP_SIZE(mon->u.pid); } else if (!is_nil(mon->name)) { /* internal by name */ - micp->mi[micp->mi_i].entity = mon->name; + micp->mi[micp->mi_i].entity.term = mon->name; micp->mi[micp->mi_i].node = erts_this_dist_entry->sysname; micp->sz += 3; /* need one 2-tuple */ } else { /* internal by pid */ - micp->mi[micp->mi_i].entity = mon->pid; + micp->mi[micp->mi_i].entity.term = mon->u.pid; micp->mi[micp->mi_i].node = NIL; /* no additional heap space needed */ } @@ -447,7 +452,7 @@ static void collect_one_origin_monitor(ErtsMonitor *mon, void *vmicp) /* have always pid at hand, to assist with figuring out if its a port or * a process, when we monitored by name and process_info is requested. * See: erl_bif_info.c:process_info_aux section for am_monitors */ - micp->mi[micp->mi_i].pid = mon->pid; + micp->mi[micp->mi_i].pid = mon->u.pid; micp->mi_i++; micp->sz += 2 + 3; /* For a cons cell and a 2-tuple */ @@ -457,15 +462,24 @@ static void collect_one_target_monitor(ErtsMonitor *mon, void *vmicp) { MonitorInfoCollection *micp = vmicp; - if (mon->type != MON_TARGET) { - return; + if (mon->type != MON_TARGET && mon->type != MON_NIF_TARGET) { + return; } EXTEND_MONITOR_INFOS(micp); - micp->mi[micp->mi_i].node = NIL; - micp->mi[micp->mi_i].entity = mon->pid; - micp->sz += (NC_HEAP_SIZE(mon->pid) + 2 /* cons */); + + if (mon->type == MON_NIF_TARGET) { + micp->mi[micp->mi_i].entity.resource = mon->u.resource; + micp->mi[micp->mi_i].node = make_small(MON_NIF_TARGET); + micp->sz += erts_resource_ref_size(mon->u.resource); + } + else { + micp->mi[micp->mi_i].entity.term = mon->u.pid; + micp->mi[micp->mi_i].node = NIL; + micp->sz += NC_HEAP_SIZE(mon->u.pid); + } + micp->sz += 2; /* cons */; micp->mi_i++; } @@ -1194,7 +1208,7 @@ process_info_aux(Process *BIF_P, hp = HAlloc(BIF_P, 3 + mic.sz); res = NIL; for (i = 0; i < mic.mi_i; i++) { - item = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity); + item = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity.term); res = CONS(hp, item, res); hp += 2; } @@ -1212,7 +1226,7 @@ process_info_aux(Process *BIF_P, hp = HAlloc(BIF_P, 3 + mic.sz); res = NIL; for (i = 0; i < mic.mi_i; i++) { - if (is_atom(mic.mi[i].entity)) { + if (is_atom(mic.mi[i].entity.term)) { /* Monitor by name. * Build {process|port, {Name, Node}} and cons it. */ @@ -1224,7 +1238,7 @@ process_info_aux(Process *BIF_P, || is_port(mic.mi[i].pid) || is_atom(mic.mi[i].pid)); - t1 = TUPLE2(hp, mic.mi[i].entity, mic.mi[i].node); + t1 = TUPLE2(hp, mic.mi[i].entity.term, mic.mi[i].node); hp += 3; t2 = TUPLE2(hp, m_type, t1); hp += 3; @@ -1234,7 +1248,7 @@ process_info_aux(Process *BIF_P, else { /* Monitor by pid. Build {process|port, Pid} and cons it. */ Eterm t; - Eterm pid = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity); + Eterm pid = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity.term); Eterm m_type = is_port(mic.mi[i].pid) ? am_port : am_process; ASSERT(is_pid(mic.mi[i].pid) @@ -1261,7 +1275,12 @@ process_info_aux(Process *BIF_P, res = NIL; for (i = 0; i < mic.mi_i; ++i) { - item = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity); + if (mic.mi[i].node == make_small(MON_NIF_TARGET)) { + item = erts_bld_resource_ref(&hp, &MSO(BIF_P), mic.mi[i].entity.resource); + } + else { + item = STORE_NC(&hp, &MSO(BIF_P), mic.mi[i].entity.term); + } res = CONS(hp, item, res); hp += 2; } @@ -2950,7 +2969,7 @@ erts_bld_port_info(Eterm **hpp, ErlOffHeap *ohp, Uint *szp, Port *prt, if (hpp) { res = NIL; for (i = 0; i < mic.mi_i; i++) { - item = STORE_NC(hpp, ohp, mic.mi[i].entity); + item = STORE_NC(hpp, ohp, mic.mi[i].entity.term); res = CONS(*hpp, item, res); *hpp += 2; } @@ -2981,7 +3000,7 @@ erts_bld_port_info(Eterm **hpp, ErlOffHeap *ohp, Uint *szp, Port *prt, Eterm t; Eterm m_type; - item = STORE_NC(hpp, ohp, mic.mi[i].entity); + item = STORE_NC(hpp, ohp, mic.mi[i].entity.term); m_type = is_port(item) ? am_port : am_process; t = TUPLE2(*hpp, m_type, item); *hpp += 3; @@ -3010,7 +3029,8 @@ erts_bld_port_info(Eterm **hpp, ErlOffHeap *ohp, Uint *szp, Port *prt, if (hpp) { res = NIL; for (i = 0; i < mic.mi_i; ++i) { - item = STORE_NC(hpp, ohp, mic.mi[i].entity); + ASSERT(mic.mi[i].node == NIL); + item = STORE_NC(hpp, ohp, mic.mi[i].entity.term); res = CONS(*hpp, item, res); *hpp += 2; } diff --git a/erts/emulator/beam/erl_bits.c b/erts/emulator/beam/erl_bits.c index 6bf52fb303..885e955332 100644 --- a/erts/emulator/beam/erl_bits.c +++ b/erts/emulator/beam/erl_bits.c @@ -110,9 +110,6 @@ erts_init_bits(void) { ERTS_CT_ASSERT(offsetof(Binary,orig_bytes) % 8 == 0); ERTS_CT_ASSERT(offsetof(ErtsMagicBinary,u.aligned.data) % 8 == 0); - ERTS_CT_ASSERT(ERTS_MAGIC_BIN_BYTES_TO_ALIGN == - (offsetof(ErtsMagicBinary,u.aligned.data) - - offsetof(ErtsMagicBinary,u.unaligned.data))); ERTS_CT_ASSERT(offsetof(ErtsBinary,driver.binary.orig_bytes) == offsetof(Binary,orig_bytes)); diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h index 5bea92e198..b386b68cff 100644 --- a/erts/emulator/beam/erl_driver.h +++ b/erts/emulator/beam/erl_driver.h @@ -175,13 +175,6 @@ struct erl_drv_event_data { #endif typedef struct erl_drv_event_data *ErlDrvEventData; /* Event data */ -/* - * A driver monitor - */ -typedef struct { - unsigned char data[sizeof(void *)*4]; -} ErlDrvMonitor; - typedef struct { unsigned long megasecs; unsigned long secs; diff --git a/erts/emulator/beam/erl_drv_nif.h b/erts/emulator/beam/erl_drv_nif.h index 46bb06d642..8de4a7855d 100644 --- a/erts/emulator/beam/erl_drv_nif.h +++ b/erts/emulator/beam/erl_drv_nif.h @@ -64,6 +64,14 @@ enum ErlNifSelectReturn { ERL_NIF_SELECT_FAILED = (1 << 4) }; +/* + * A driver monitor + */ +typedef struct { + unsigned char data[sizeof(void *)*4]; +} ErlDrvMonitor; + + #ifdef SIZEOF_CHAR # define SIZEOF_CHAR_SAVED__ SIZEOF_CHAR # undef SIZEOF_CHAR diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c index 08dcbed91c..0798faf53a 100644 --- a/erts/emulator/beam/erl_lock_check.c +++ b/erts/emulator/beam/erl_lock_check.c @@ -89,6 +89,9 @@ static erts_lc_lock_order_t erts_lock_order[] = { { "hipe_mfait_lock", NULL }, #endif { "nodes_monitors", NULL }, +#ifdef ERTS_SMP + { "resource_monitors", "address" }, +#endif { "driver_list", NULL }, { "proc_link", "pid" }, { "proc_msgq", "pid" }, diff --git a/erts/emulator/beam/erl_monitors.c b/erts/emulator/beam/erl_monitors.c index 910598690d..e5be66be79 100644 --- a/erts/emulator/beam/erl_monitors.c +++ b/erts/emulator/beam/erl_monitors.c @@ -91,7 +91,7 @@ static ERTS_INLINE int cmp_mon_ref(Eterm ref1, Eterm ref2) #define CP_LINK_VAL(To, Hp, From) \ do { \ - if (IS_CONST(From)) \ + if (is_immed(From)) \ (To) = (From); \ else { \ Uint i__; \ @@ -109,15 +109,15 @@ do { \ } \ } while (0) -static ErtsMonitor *create_monitor(Uint type, Eterm ref, Eterm pid, Eterm name) +static ErtsMonitor *create_monitor(Uint type, Eterm ref, UWord entity, Eterm name) { Uint mon_size = ERTS_MONITOR_SIZE; ErtsMonitor *n; Eterm *hp; mon_size += NC_HEAP_SIZE(ref); - if (!IS_CONST(pid)) { - mon_size += NC_HEAP_SIZE(pid); + if (type != MON_NIF_TARGET && is_not_immed(entity)) { + mon_size += NC_HEAP_SIZE(entity); } if (mon_size <= ERTS_MONITOR_SH_SIZE) { @@ -136,7 +136,10 @@ static ErtsMonitor *create_monitor(Uint type, Eterm ref, Eterm pid, Eterm name) n->balance = 0; /* Always the same initial value */ n->name = name; /* atom() or [] */ CP_LINK_VAL(n->ref, hp, ref); /*XXX Unneccesary check, never immediate*/ - CP_LINK_VAL(n->pid, hp, pid); + if (type == MON_NIF_TARGET) + n->u.resource = (ErtsResource*)entity; + else + CP_LINK_VAL(n->u.pid, hp, (Eterm)entity); return n; } @@ -147,7 +150,7 @@ static ErtsLink *create_link(Uint type, Eterm pid) ErtsLink *n; Eterm *hp; - if (!IS_CONST(pid)) { + if (is_not_immed(pid)) { lnk_size += NC_HEAP_SIZE(pid); } @@ -206,16 +209,16 @@ void erts_destroy_monitor(ErtsMonitor *mon) Uint mon_size = ERTS_MONITOR_SIZE; ErlNode *node; - ASSERT(!IS_CONST(mon->ref)); + ASSERT(is_not_immed(mon->ref)); mon_size += NC_HEAP_SIZE(mon->ref); if (is_external(mon->ref)) { node = external_thing_ptr(mon->ref)->node; erts_deref_node_entry(node); } - if (!IS_CONST(mon->pid)) { - mon_size += NC_HEAP_SIZE(mon->pid); - if (is_external(mon->pid)) { - node = external_thing_ptr(mon->pid)->node; + if (mon->type != MON_NIF_TARGET && is_not_immed(mon->u.pid)) { + mon_size += NC_HEAP_SIZE(mon->u.pid); + if (is_external(mon->u.pid)) { + node = external_thing_ptr(mon->u.pid)->node; erts_deref_node_entry(node); } } @@ -234,7 +237,7 @@ void erts_destroy_link(ErtsLink *lnk) ASSERT(lnk->type == LINK_NODE || ERTS_LINK_ROOT(lnk) == NULL); - if (!IS_CONST(lnk->pid)) { + if (is_not_immed(lnk->pid)) { lnk_size += NC_HEAP_SIZE(lnk->pid); if (is_external(lnk->pid)) { node = external_thing_ptr(lnk->pid)->node; @@ -329,7 +332,7 @@ static void insertion_rotation(int dstack[], int dpos, } } -void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, Eterm pid, +void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, UWord entity, Eterm name) { void *tstack[STACK_NEED]; @@ -344,7 +347,7 @@ void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, Eterm pid, for (;;) { if (!*this) { /* Found our place */ state = 1; - *this = create_monitor(type,ref,pid,name); + *this = create_monitor(type,ref,entity,name); break; } else if ((c = CMP_MON_REF(ref,(*this)->ref)) < 0) { /* go left */ @@ -914,8 +917,12 @@ static void erts_dump_monitors(ErtsMonitor *root, int indent) if (root == NULL) return; erts_dump_monitors(root->right,indent+2); - erts_printf("%*s[%b16d:%b16u:%T:%T:%T]\n", indent, "", root->balance, - root->type, root->ref, root->pid, root->name); + erts_printf("%*s[%b16d:%b16u:%T:%T", indent, "", root->balance, + root->type, root->ref, root->name); + if (root->type == MON_NIF_TARGET) + erts_printf(":%p]\n", root->u.resource); + else + erts_printf(":%T]\n", root->u.pid); erts_dump_monitors(root->left,indent+2); } @@ -1030,7 +1037,7 @@ void erts_one_link_size(ErtsLink *lnk, void *vpu) { Uint *pu = vpu; *pu += ERTS_LINK_SIZE*sizeof(Uint); - if(!IS_CONST(lnk->pid)) + if(is_not_immed(lnk->pid)) *pu += NC_HEAP_SIZE(lnk->pid)*sizeof(Uint); if (lnk->type != LINK_NODE && ERTS_LINK_ROOT(lnk) != NULL) { erts_doforall_links(ERTS_LINK_ROOT(lnk),&erts_one_link_size,vpu); @@ -1040,8 +1047,8 @@ void erts_one_mon_size(ErtsMonitor *mon, void *vpu) { Uint *pu = vpu; *pu += ERTS_MONITOR_SIZE*sizeof(Uint); - if(!IS_CONST(mon->pid)) - *pu += NC_HEAP_SIZE(mon->pid)*sizeof(Uint); - if(!IS_CONST(mon->ref)) + if(mon->type != MON_NIF_TARGET && is_not_immed(mon->u.pid)) + *pu += NC_HEAP_SIZE(mon->u.pid)*sizeof(Uint); + if(is_not_immed(mon->ref)) *pu += NC_HEAP_SIZE(mon->ref)*sizeof(Uint); } diff --git a/erts/emulator/beam/erl_monitors.h b/erts/emulator/beam/erl_monitors.h index 9e2beedea3..f90ac930e5 100644 --- a/erts/emulator/beam/erl_monitors.h +++ b/erts/emulator/beam/erl_monitors.h @@ -82,8 +82,9 @@ /* Type tags for monitors */ #define MON_ORIGIN 1 -#define MON_TARGET 3 -#define MON_TIME_OFFSET 7 +#define MON_TARGET 2 +#define MON_NIF_TARGET 3 +#define MON_TIME_OFFSET 4 /* Type tags for links */ #define LINK_PID 1 /* ...Or port */ @@ -105,11 +106,15 @@ typedef struct erts_monitor_or_link { typedef struct erts_monitor { struct erts_monitor *left, *right; Sint16 balance; - Uint16 type; /* MON_ORIGIN | MON_TARGET | MON_TIME_OFFSET */ + Uint16 type; /* MON_ORIGIN | MON_TARGET | MON_NIF_TARGET | MON_TIME_OFFSET */ Eterm ref; - Eterm pid; /* In case of distributed named monitor, this is the - nodename atom in MON_ORIGIN process, otherwise a pid or - , in case of a MON_TARGET, a port */ + union { + Eterm pid; /* In case of distributed named monitor, this is the + * nodename atom in MON_ORIGIN process, otherwise a pid or, + * in case of a MON_TARGET, a port + */ + struct ErtsResource_* resource; /* MON_NIF_TARGET */ + }u; Eterm name; /* When monitoring a named process: atom() else [] */ Uint heap[1]; /* Larger in reality */ } ErtsMonitor; @@ -144,7 +149,7 @@ Uint erts_tot_link_lh_size(void); /* Prototypes */ void erts_destroy_monitor(ErtsMonitor *mon); -void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, Eterm pid, +void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, UWord entity, Eterm name); ErtsMonitor *erts_remove_monitor(ErtsMonitor **root, Eterm ref); ErtsMonitor *erts_lookup_monitor(ErtsMonitor *root, Eterm ref); diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c index b7425b9e45..62191c4abf 100644 --- a/erts/emulator/beam/erl_nif.c +++ b/erts/emulator/beam/erl_nif.c @@ -1913,8 +1913,6 @@ int enif_snprintf(char *buffer, size_t size, const char* format, ...) /* dummy node in circular list */ struct enif_resource_type_t resource_type_list; -#define SIZEOF_ErlNifResource(SIZE) (offsetof(ErtsResource,data) + (SIZE)) - static ErlNifResourceType* find_resource_type(Eterm module, Eterm name) { ErlNifResourceType* type; @@ -2083,6 +2081,7 @@ static void commit_opened_resource_types(struct erl_module_nif* lib) type->owner = lib; type->dtor = ort->new_callbacks.dtor; type->stop = ort->new_callbacks.stop; + type->down = ort->new_callbacks.down; if (type->dtor != NULL) { erts_refc_inc(&lib->rt_dtor_cnt, 1); @@ -2108,12 +2107,148 @@ static void rollback_opened_resource_types(void) } } +struct destroy_monitor_ctx +{ + Binary* resource_bin; + int exiting_procs; + int scheduler; +}; + +static void destroy_one_monitor(ErtsMonitor* mon, void* context) +{ + struct destroy_monitor_ctx* ctx = (struct destroy_monitor_ctx*) context; + Process* rp; + ErtsMonitor *rmon = NULL; + int is_exiting; + + ASSERT(mon->type == MON_ORIGIN); + ASSERT(is_internal_pid(mon->u.pid)); + ASSERT(is_internal_ref(mon->ref)); + + if (ctx->scheduler > 0) { /* Normal scheduler */ + rp = erts_proc_lookup(mon->u.pid); + } + else { +#ifdef ERTS_SMP + rp = erts_proc_lookup_inc_refc(mon->u.pid); +#else + ASSERT(!"nif monitor destruction in non-scheduler thread"); + rp = NULL; +#endif + } + + if (!rp) { + is_exiting = 1; + } + if (rp) { + erts_smp_proc_lock(rp, ERTS_PROC_LOCK_LINK); + if (ERTS_PROC_IS_EXITING(rp)) { + is_exiting = 1; + } else { + rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref); + ASSERT(rmon); + is_exiting = 0; + } + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); +#ifdef ERTS_SMP + if (ctx->scheduler <= 0) + erts_proc_dec_refc(rp); +#endif + } + if (is_exiting) { + ctx->exiting_procs++; + /* +1 for exiting process */ + erts_refc_inc(&ctx->resource_bin->refc, ctx->exiting_procs); + } + + /* ToDo: Delay destruction after monitor_locks */ + if (rmon) { + ASSERT(rmon->type == MON_NIF_TARGET); + ASSERT(&ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(rmon->u.pid)->binary == ctx->resource_bin); + erts_destroy_monitor(rmon); + } + erts_destroy_monitor(mon); +} + +static int destroy_all_monitors(ErtsMonitor* monitors, Binary* bin) +{ + struct destroy_monitor_ctx ctx; + + execution_state(NULL, NULL, &ctx.scheduler); + + ctx.resource_bin = bin; + ctx.exiting_procs = 0; + erts_sweep_monitors(monitors, &destroy_one_monitor, &ctx); + return ctx.exiting_procs == 0; +} + + +#ifdef ERTS_SMP +# define NIF_RESOURCE_DTOR &nif_resource_dtor +#else +# define NIF_RESOURCE_DTOR &nosmp_nif_resource_dtor_prologue + +/* + * NO-SMP: Always run resource destructor on scheduler thread + * as we may have to remove process monitors. + */ +static int nif_resource_dtor(Binary*); + +static void nosmp_nif_resource_dtor_scheduled(void* vbin) +{ + erts_bin_free((Binary*)vbin); +} + +static int nosmp_nif_resource_dtor_prologue(Binary* bin) +{ + if (is_scheduler()) { + return nif_resource_dtor(bin); + } + else { + erts_schedule_misc_aux_work(1, nosmp_nif_resource_dtor_scheduled, bin); + return 0; /* do not free */ + } +} + +#endif /* !ERTS_SMP */ static int nif_resource_dtor(Binary* bin) { ErtsResource* resource = (ErtsResource*) ERTS_MAGIC_BIN_UNALIGNED_DATA(bin); ErlNifResourceType* type = resource->type; - ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == &nif_resource_dtor); + ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == NIF_RESOURCE_DTOR); + + if (resource->monitors) { + ErtsResourceMonitors* rm = resource->monitors; + ErtsMonitor* root; + + ASSERT(type->down); + erts_smp_mtx_lock(&rm->lock); + ASSERT(erts_refc_read(&bin->refc, 0) == 0); + root = rm->root; + if (root) { + rm->root = NULL; + if (!destroy_all_monitors(root, bin)) { + /* + * Resource death struggle prolonged to serve exiting process(es). + * Destructor will be called again when last exiting process + * tries to fire its MON_NIF_TARGET monitor (and fails). + * + * This resource is doomed. It has no "real" references and + * should get not get called upon to do anything except the + * final destructor call. + */ + ASSERT(erts_refc_read(&bin->refc, 1)); +#ifdef DEBUG + resource->dbg_is_dying = 1; +#endif + erts_smp_mtx_unlock(&rm->lock); + return 0; + } + } + erts_smp_mtx_unlock(&rm->lock); + erts_smp_mtx_destroy(&rm->lock); + } if (type->dtor != NULL) { struct enif_msg_environment_t msg_env; @@ -2141,20 +2276,99 @@ void erts_resource_stop(ErtsResource* resource, ErlNifEvent e, post_nif_noproc(&msg_env); } -void* enif_alloc_resource(ErlNifResourceType* type, size_t size) +/* SVERK SVERK: Move to ?.h */ +void erts_ref_to_driver_monitor(Eterm ref, ErlDrvMonitor*); + +void erts_fire_nif_monitor(ErtsResource* resource, Eterm pid, Eterm ref) { - Binary* bin = erts_create_magic_binary_x(SIZEOF_ErlNifResource(size), - &nif_resource_dtor, - 1); /* unaligned */ - ErtsResource* resource = ERTS_MAGIC_BIN_UNALIGNED_DATA(bin); + ErtsMonitor* rmon; + ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); + struct enif_msg_environment_t msg_env; + ErlNifPid nif_pid; + ErlNifMonitor nif_monitor; + ErtsResourceMonitors* rmp = resource->monitors; + + ASSERT(rmp); + ASSERT(resource->type->down); + + erts_smp_mtx_lock(&rmp->lock); + rmon = erts_remove_monitor(&rmp->root, ref); + if (!rmon) { + erts_smp_mtx_unlock(&rmp->lock); + /* -1 for exiting process */ + if (erts_refc_dectest(&bin->binary.refc, 0) == 0) { + erts_bin_free(&bin->binary); + } + return; + } + ASSERT(!resource->dbg_is_dying); + if (erts_refc_inctest(&bin->binary.refc, 1) < 2) { + /* + * Racing resource destruction. + * To avoid a more complex refc-dance with destructing thread + * we avoid calling 'down' and just silently remove the monitor. + * There are no real references left to this resource and we have + * monitors_lock, so it's safe to reset refc back to zero. + * This can happen even for non smp as destructor calls may be scheduled. + */ + erts_refc_init(&bin->binary.refc, 0); + erts_smp_mtx_unlock(&rmp->lock); + } + else { + erts_smp_mtx_unlock(&rmp->lock); + + ASSERT(rmon->u.pid == pid); + erts_ref_to_driver_monitor(ref, &nif_monitor); + nif_pid.pid = pid; + pre_nif_noproc(&msg_env, resource->type->owner, NULL); + resource->type->down(&msg_env.env, resource->data, &nif_pid, &nif_monitor); + post_nif_noproc(&msg_env); + + if (erts_refc_dectest(&bin->binary.refc, 0) == 0) { + erts_bin_free(&bin->binary); + } + } + erts_destroy_monitor(rmon); +} + +void* enif_alloc_resource(ErlNifResourceType* type, size_t data_sz) +{ + size_t magic_sz = offsetof(ErtsResource,data); + Binary* bin; + ErtsResource* resource; + size_t monitors_offs; + + if (type->down) { + /* Put ErtsResourceMonitors after user data and properly aligned */ + monitors_offs = ((data_sz + ERTS_ALLOC_ALIGN_BYTES - 1) + & ~((size_t)ERTS_ALLOC_ALIGN_BYTES - 1)); + magic_sz += monitors_offs + sizeof(ErtsResourceMonitors); + } + else { + ERTS_UNDEF(monitors_offs, 0); + magic_sz += data_sz; + } + bin = erts_create_magic_binary_x(magic_sz, NIF_RESOURCE_DTOR, + 1); /* unaligned */ + resource = ERTS_MAGIC_BIN_UNALIGNED_DATA(bin); ASSERT(type->owner && type->next && type->prev); /* not allowed in load/upgrade */ resource->type = type; erts_refc_inc(&bin->refc, 1); #ifdef DEBUG erts_refc_init(&resource->nif_refc, 1); + resource->dbg_is_dying = 0; #endif erts_refc_inc(&resource->type->refc, 2); + if (type->down) { + resource->monitors = (ErtsResourceMonitors*) (resource->data + monitors_offs); + erts_smp_mtx_init(&resource->monitors->lock, "resource_monitors"); + resource->monitors->root = NULL; + resource->monitors->user_data_sz = data_sz; + } + else { + resource->monitors = NULL; + } return resource->data; } @@ -2163,7 +2377,8 @@ void enif_release_resource(void* obj) ErtsResource* resource = DATA_TO_RESOURCE(obj); ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); - ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == &nif_resource_dtor); + ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == NIF_RESOURCE_DTOR); + ASSERT(!resource->dbg_is_dying); #ifdef DEBUG erts_refc_dec(&resource->nif_refc, 0); #endif @@ -2177,18 +2392,27 @@ void enif_keep_resource(void* obj) ErtsResource* resource = DATA_TO_RESOURCE(obj); ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); - ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == &nif_resource_dtor); + ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == NIF_RESOURCE_DTOR); + ASSERT(!resource->dbg_is_dying); #ifdef DEBUG erts_refc_inc(&resource->nif_refc, 1); #endif erts_refc_inc(&bin->binary.refc, 2); } +Eterm erts_bld_resource_ref(Eterm** hpp, ErlOffHeap* oh, ErtsResource* resource) +{ + ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); + ASSERT(!resource->dbg_is_dying); + return erts_mk_magic_binary_term(hpp, oh, &bin->binary); +} + ERL_NIF_TERM enif_make_resource(ErlNifEnv* env, void* obj) { ErtsResource* resource = DATA_TO_RESOURCE(obj); ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); Eterm* hp = alloc_heap(env,PROC_BIN_SIZE); + ASSERT(!resource->dbg_is_dying); return erts_mk_magic_binary_term(&hp, &MSO(env->proc), &bin->binary); } @@ -2217,7 +2441,7 @@ int enif_get_resource(ErlNifEnv* env, ERL_NIF_TERM term, ErlNifResourceType* typ }*/ mbin = pb->val; resource = (ErtsResource*) ERTS_MAGIC_BIN_UNALIGNED_DATA(mbin); - if (ERTS_MAGIC_BIN_DESTRUCTOR(mbin) != &nif_resource_dtor + if (ERTS_MAGIC_BIN_DESTRUCTOR(mbin) != NIF_RESOURCE_DTOR || resource->type != type) { return 0; } @@ -2228,8 +2452,13 @@ int enif_get_resource(ErlNifEnv* env, ERL_NIF_TERM term, ErlNifResourceType* typ size_t enif_sizeof_resource(void* obj) { ErtsResource* resource = DATA_TO_RESOURCE(obj); - Binary* bin = &ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource)->binary; - return ERTS_MAGIC_BIN_UNALIGNED_DATA_SIZE(bin) - offsetof(ErtsResource,data); + if (resource->monitors) { + return resource->monitors->user_data_sz; + } + else { + Binary* bin = &ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource)->binary; + return ERTS_MAGIC_BIN_UNALIGNED_DATA_SIZE(bin) - offsetof(ErtsResource,data); + } } @@ -2830,6 +3059,150 @@ int enif_map_iterator_get_pair(ErlNifEnv *env, return 0; } +int enif_monitor_process(ErlNifEnv* env, void* obj, const ErlNifPid* target_pid, + ErlNifMonitor* monitor) +{ + int scheduler; + ErtsResource* rsrc = DATA_TO_RESOURCE(obj); + Process *rp; + Eterm tmp[REF_THING_SIZE]; + Eterm ref; + int retval; + + ASSERT(ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(rsrc)->magic_binary.destructor + == NIF_RESOURCE_DTOR); + ASSERT(!rsrc->dbg_is_dying); + ASSERT(!rsrc->monitors == !rsrc->type->down); + + + if (!rsrc->monitors) { + ASSERT(!rsrc->type->down); + return -1; + } + ASSERT(rsrc->type->down); + + execution_state(env, NULL, &scheduler); + +#ifdef ERTS_SMP + if (scheduler > 0) /* Normal scheduler */ + rp = erts_proc_lookup_raw(target_pid->pid); + else + rp = erts_proc_lookup_raw_inc_refc(target_pid->pid); +#else + if (scheduler <= 0) { + erts_exit(ERTS_ABORT_EXIT, "enif_monitor_process: called from " + "non-scheduler thread on non-SMP VM"); + } + rp = erts_proc_lookup(target_pid->pid); +#endif + + if (!rp) + return 1; + + ref = erts_make_ref_in_buffer(tmp); + + erts_smp_mtx_lock(&rsrc->monitors->lock); + erts_smp_proc_lock(rp, ERTS_PROC_LOCK_LINK); + if (ERTS_PSFLG_FREE & erts_smp_atomic32_read_nob(&rp->state)) { + retval = 1; + } + else { + erts_add_monitor(&rsrc->monitors->root, MON_ORIGIN, ref, rp->common.id, NIL); + erts_add_monitor(&ERTS_P_MONITORS(rp), MON_NIF_TARGET, ref, (UWord)rsrc, NIL); + retval = 0; + } + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); + erts_smp_mtx_unlock(&rsrc->monitors->lock); + +#ifdef ERTS_SMP + if (scheduler <= 0) + erts_proc_dec_refc(rp); +#endif + if (monitor) + erts_ref_to_driver_monitor(ref,monitor); + + return retval; +} + +int enif_demonitor_process(ErlNifEnv* env, void* obj, const ErlNifMonitor* monitor) +{ + int scheduler; + ErtsResource* rsrc = DATA_TO_RESOURCE(obj); + ErtsBinary* bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(rsrc); + Process *rp; + ErtsMonitor *mon; + ErtsMonitor *rmon = NULL; + Eterm ref_heap[REF_THING_SIZE]; + Eterm ref; + int is_exiting; + + ASSERT(bin->magic_binary.destructor == NIF_RESOURCE_DTOR); + ASSERT(!rsrc->dbg_is_dying); + + execution_state(env, NULL, &scheduler); + + memcpy(ref_heap, monitor, sizeof(Eterm)*REF_THING_SIZE); + ref = make_internal_ref(ref_heap); + erts_smp_mtx_lock(&rsrc->monitors->lock); + mon = erts_remove_monitor(&rsrc->monitors->root, ref); + + if (mon == NULL) { + erts_smp_mtx_unlock(&rsrc->monitors->lock); + return 1; + } + + ASSERT(mon->type == MON_ORIGIN); + ASSERT(is_internal_pid(mon->u.pid)); + +#ifdef ERTS_SMP + if (scheduler > 0) /* Normal scheduler */ + rp = erts_proc_lookup(mon->u.pid); + else + rp = erts_proc_lookup_inc_refc(mon->u.pid); +#else + if (scheduler <= 0) { + erts_exit(ERTS_ABORT_EXIT, "enif_demonitor_process: called from " + "non-scheduler thread on non-SMP VM"); + } + rp = erts_proc_lookup(mon->u.pid); +#endif + + if (!rp) { + is_exiting = 1; + } + else { + erts_smp_proc_lock(rp, ERTS_PROC_LOCK_LINK); + if (ERTS_PROC_IS_EXITING(rp)) { + is_exiting = 1; + } else { + rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), ref); + ASSERT(rmon); + is_exiting = 0; + } + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); + +#ifdef ERTS_SMP + if (scheduler <= 0) + erts_proc_dec_refc(rp); +#endif + } + if (is_exiting) { + /* +1 for exiting process */ + erts_refc_inc(&bin->binary.refc, 2); + } + erts_smp_mtx_unlock(&rsrc->monitors->lock); + + if (rmon) { + ASSERT(rmon->type == MON_NIF_TARGET); + ASSERT(rmon->u.resource == rsrc); + erts_destroy_monitor(rmon); + } + erts_destroy_monitor(mon); + + return 0; +} + + /*************************************************************************** ** load_nif/2 ** ***************************************************************************/ @@ -3303,7 +3676,8 @@ erts_unload_nif(struct erl_module_nif* lib) void erl_nif_init() { - ERTS_CT_ASSERT((offsetof(ErtsResource,data) % 8) == ERTS_MAGIC_BIN_BYTES_TO_ALIGN); + ERTS_CT_ASSERT((offsetof(ErtsResource,data) % 8) + == ERTS_MAGIC_BIN_BYTES_TO_ALIGN); resource_type_list.next = &resource_type_list; resource_type_list.prev = &resource_type_list; diff --git a/erts/emulator/beam/erl_nif.h b/erts/emulator/beam/erl_nif.h index 78e0fa1864..5248f287ee 100644 --- a/erts/emulator/beam/erl_nif.h +++ b/erts/emulator/beam/erl_nif.h @@ -142,15 +142,6 @@ typedef struct typedef int ErlNifEvent; /* An event to be selected on. */ //#endif -typedef struct enif_resource_type_t ErlNifResourceType; -typedef void ErlNifResourceDtor(ErlNifEnv*, void*); -typedef void ErlNifResourceStop(ErlNifEnv*, void*, ErlNifEvent, int is_direct_call); - -typedef struct { - ErlNifResourceDtor* dtor; - ErlNifResourceStop* stop; /* at ERL_NIF_SELECT_STOP event */ -} ErlNifResourceTypeInit; - typedef enum { ERL_NIF_RT_CREATE = 1, @@ -172,6 +163,19 @@ typedef struct ERL_NIF_TERM port_id; /* internal, may change */ }ErlNifPort; +typedef ErlDrvMonitor ErlNifMonitor; + +typedef struct enif_resource_type_t ErlNifResourceType; +typedef void ErlNifResourceDtor(ErlNifEnv*, void*); +typedef void ErlNifResourceStop(ErlNifEnv*, void*, ErlNifEvent, int is_direct_call); +typedef void ErlNifResourceDown(ErlNifEnv*, void*, ErlNifPid*, ErlNifMonitor*); + +typedef struct { + ErlNifResourceDtor* dtor; + ErlNifResourceStop* stop; /* at ERL_NIF_SELECT_STOP event */ + ErlNifResourceDown* down; /* enif_monitor_process */ +} ErlNifResourceTypeInit; + typedef ErlDrvSysInfo ErlNifSysInfo; typedef struct ErlDrvTid_ *ErlNifTid; diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h index 9163ce25eb..a9ed246962 100644 --- a/erts/emulator/beam/erl_nif_api_funcs.h +++ b/erts/emulator/beam/erl_nif_api_funcs.h @@ -177,6 +177,8 @@ ERL_NIF_API_FUNC_DECL(int,enif_thread_type,(void)); ERL_NIF_API_FUNC_DECL(int,enif_snprintf,(char * buffer, size_t size, const char *format, ...)); ERL_NIF_API_FUNC_DECL(int,enif_select,(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags flags, void* obj, ERL_NIF_TERM ref)); ERL_NIF_API_FUNC_DECL(ErlNifResourceType*,enif_open_resource_type_x,(ErlNifEnv*, const char* name_str, const ErlNifResourceTypeInit*, ErlNifResourceFlags flags, ErlNifResourceFlags* tried)); +ERL_NIF_API_FUNC_DECL(int, enif_monitor_process,(ErlNifEnv*,void* obj,const ErlNifPid*,ErlDrvMonitor *monitor)); +ERL_NIF_API_FUNC_DECL(int, enif_demonitor_process,(ErlNifEnv*,void* obj,const ErlDrvMonitor *monitor)); /* ** ADD NEW ENTRIES HERE (before this comment) !!! @@ -336,6 +338,8 @@ ERL_NIF_API_FUNC_DECL(ErlNifResourceType*,enif_open_resource_type_x,(ErlNifEnv*, # define enif_snprintf ERL_NIF_API_FUNC_MACRO(enif_snprintf) # define enif_select ERL_NIF_API_FUNC_MACRO(enif_select) # define enif_open_resource_type_x ERL_NIF_API_FUNC_MACRO(enif_open_resource_type_x) +# define enif_monitor_process ERL_NIF_API_FUNC_MACRO(enif_monitor_process) +# define enif_demonitor_process ERL_NIF_API_FUNC_MACRO(enif_demonitor_process) /* ** ADD NEW ENTRIES HERE (before this comment) diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c index 70500ed6e1..f463f7fdf4 100644 --- a/erts/emulator/beam/erl_node_tables.c +++ b/erts/emulator/beam/erl_node_tables.c @@ -1165,8 +1165,8 @@ insert_offheap(ErlOffHeap *oh, int type, Eterm id) static void doit_insert_monitor(ErtsMonitor *monitor, void *p) { Eterm *idp = p; - if(is_external(monitor->pid)) - insert_node(external_thing_ptr(monitor->pid)->node, MONITOR_REF, *idp); + if(monitor->type != MON_NIF_TARGET && is_external(monitor->u.pid)) + insert_node(external_thing_ptr(monitor->u.pid)->node, MONITOR_REF, *idp); if(is_external(monitor->ref)) insert_node(external_thing_ptr(monitor->ref)->node, MONITOR_REF, *idp); } diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c index f80ebdf31b..9db77585f2 100644 --- a/erts/emulator/beam/erl_process.c +++ b/erts/emulator/beam/erl_process.c @@ -13320,9 +13320,9 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) switch (mon->type) { case MON_ORIGIN: /* We are monitoring someone else, we need to demonitor that one.. */ - if (is_atom(mon->pid)) { /* remote by name */ - ASSERT(is_node_name_atom(mon->pid)); - dep = erts_sysname_to_connected_dist_entry(mon->pid); + if (is_atom(mon->u.pid)) { /* remote by name */ + ASSERT(is_node_name_atom(mon->u.pid)); + dep = erts_sysname_to_connected_dist_entry(mon->u.pid); if (dep) { erts_smp_de_links_lock(dep); rmon = erts_remove_monitor(&(dep->monitors), mon->ref); @@ -13333,7 +13333,7 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) ERTS_DSP_NO_LOCK, 0); if (code == ERTS_DSIG_PREP_CONNECTED) { code = erts_dsig_send_demonitor(&dsd, - rmon->pid, + rmon->u.pid, mon->name, mon->ref, 1); @@ -13344,10 +13344,10 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) erts_deref_dist_entry(dep); } } else { - ASSERT(is_pid(mon->pid) || is_port(mon->pid)); + ASSERT(is_pid(mon->u.pid) || is_port(mon->u.pid)); /* if is local by pid or name */ - if (is_internal_pid(mon->pid)) { - Process *rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK); + if (is_internal_pid(mon->u.pid)) { + Process *rp = erts_pid2proc(NULL, 0, mon->u.pid, ERTS_PROC_LOCK_LINK); if (!rp) { goto done; } @@ -13357,9 +13357,9 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) goto done; } erts_destroy_monitor(rmon); - } else if (is_internal_port(mon->pid)) { + } else if (is_internal_port(mon->u.pid)) { /* Is a local port */ - Port *prt = erts_port_lookup_raw(mon->pid); + Port *prt = erts_port_lookup_raw(mon->u.pid); if (!prt) { goto done; } @@ -13367,8 +13367,8 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) ERTS_PORT_DEMONITOR_ORIGIN_ON_DEATHBED, prt, mon->ref, NULL); } else { /* remote by pid */ - ASSERT(is_external_pid(mon->pid)); - dep = external_pid_dist_entry(mon->pid); + ASSERT(is_external_pid(mon->u.pid)); + dep = external_pid_dist_entry(mon->u.pid); ASSERT(dep != NULL); if (dep) { erts_smp_de_links_lock(dep); @@ -13380,8 +13380,8 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) ERTS_DSP_NO_LOCK, 0); if (code == ERTS_DSIG_PREP_CONNECTED) { code = erts_dsig_send_demonitor(&dsd, - rmon->pid, - mon->pid, + rmon->u.pid, + mon->u.pid, mon->ref, 1); ASSERT(code == ERTS_DSIG_SEND_OK); @@ -13393,22 +13393,21 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) } break; case MON_TARGET: - ASSERT(mon->type == MON_TARGET); - ASSERT(is_pid(mon->pid) || is_internal_port(mon->pid)); - if (is_internal_port(mon->pid)) { - Port *prt = erts_id2port(mon->pid); + ASSERT(is_pid(mon->u.pid) || is_internal_port(mon->u.pid)); + if (is_internal_port(mon->u.pid)) { + Port *prt = erts_id2port(mon->u.pid); if (prt == NULL) { goto done; } erts_fire_port_monitor(prt, mon->ref); erts_port_release(prt); - } else if (is_internal_pid(mon->pid)) {/* local by name or pid */ + } else if (is_internal_pid(mon->u.pid)) {/* local by name or pid */ Eterm watched; Process *rp; DeclareTmpHeapNoproc(lhp,3); ErtsProcLocks rp_locks = (ERTS_PROC_LOCK_LINK | ERTS_PROC_LOCKS_MSG_SEND); - rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks); + rp = erts_pid2proc(NULL, 0, mon->u.pid, rp_locks); if (rp == NULL) { goto done; } @@ -13427,8 +13426,8 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) /* else: demonitor while we exited, i.e. do nothing... */ erts_smp_proc_unlock(rp, rp_locks); } else { /* external by pid or name */ - ASSERT(is_external_pid(mon->pid)); - dep = external_pid_dist_entry(mon->pid); + ASSERT(is_external_pid(mon->u.pid)); + dep = external_pid_dist_entry(mon->u.pid); ASSERT(dep != NULL); if (dep) { erts_smp_de_links_lock(dep); @@ -13440,10 +13439,10 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) ERTS_DSP_NO_LOCK, 0); if (code == ERTS_DSIG_PREP_CONNECTED) { code = erts_dsig_send_m_exit(&dsd, - mon->pid, + mon->u.pid, (rmon->name != NIL ? rmon->name - : rmon->pid), + : rmon->u.pid), mon->ref, pcontext->reason); ASSERT(code == ERTS_DSIG_SEND_OK); @@ -13453,6 +13452,11 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) } } break; + case MON_NIF_TARGET: + erts_fire_nif_monitor(mon->u.resource, + pcontext->p->common.id, + mon->ref); + break; case MON_TIME_OFFSET: erts_demonitor_time_offset(mon->ref); break; diff --git a/erts/emulator/beam/erl_time_sup.c b/erts/emulator/beam/erl_time_sup.c index 6aa2a7500f..c01cc26246 100644 --- a/erts/emulator/beam/erl_time_sup.c +++ b/erts/emulator/beam/erl_time_sup.c @@ -1866,7 +1866,7 @@ save_time_offset_monitor(ErtsMonitor *mon, void *vcntxt) cntxt = (ErtsTimeOffsetMonitorContext *) vcntxt; mix = (cntxt->ix)++; - cntxt->to_mon_info[mix].pid = mon->pid; + cntxt->to_mon_info[mix].pid = mon->u.pid; to_hp = &cntxt->to_mon_info[mix].heap[0]; ASSERT(is_internal_ref(mon->ref)); diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h index 86c38f8e8c..511e357d14 100644 --- a/erts/emulator/beam/global.h +++ b/erts/emulator/beam/global.h @@ -66,30 +66,47 @@ struct enif_resource_type_t struct erl_module_nif* owner; /* that created this type and thus implements the destructor*/ ErlNifResourceDtor* dtor; /* user destructor function */ ErlNifResourceStop* stop; + ErlNifResourceDown* down; erts_refc_t refc; /* num of resources of this type (HOTSPOT warning) +1 for active erl_module_nif */ Eterm module; Eterm name; }; + typedef struct +{ + erts_smp_mtx_t lock; + ErtsMonitor* root; + size_t user_data_sz; +} ErtsResourceMonitors; + +typedef struct ErtsResource_ { struct enif_resource_type_t* type; + ErtsResourceMonitors* monitors; +#ifdef ARCH_32 + byte align__[4]; +#endif #ifdef DEBUG erts_refc_t nif_refc; -# ifdef ARCH_32 - byte align__[4]; + int dbg_is_dying; +# ifdef ARCH_64 + byte dbg_align__[4]; # endif #endif - char data[1]; }ErtsResource; -#define DATA_TO_RESOURCE(PTR) ((ErtsResource*)((char*)(PTR) - offsetof(ErtsResource,data))) +#define DATA_TO_RESOURCE(PTR) ErtsContainerStruct(PTR, ErtsResource, data) +#define erts_resource_ref_size(P) PROC_BIN_SIZE + +extern Eterm erts_bld_resource_ref(Eterm** hp, ErlOffHeap*, ErtsResource*); extern void erts_pre_nif(struct enif_environment_t*, Process*, struct erl_module_nif*, Process* tracee); extern void erts_post_nif(struct enif_environment_t* env); extern void erts_resource_stop(ErtsResource*, ErlNifEvent, int is_direct_call); +void erts_fire_nif_monitor(ErtsResource*, Eterm pid, Eterm ref); extern Eterm erts_nif_taints(Process* p); extern void erts_print_nif_taints(fmtfn_t to, void* to_arg); void erts_unload_nif(struct erl_module_nif* nif); @@ -286,11 +303,9 @@ typedef struct { } u; } ErtsMagicBinary; -#ifdef ARCH_32 -#define ERTS_MAGIC_BIN_BYTES_TO_ALIGN 4 -#else -#define ERTS_MAGIC_BIN_BYTES_TO_ALIGN 0 -#endif +#define ERTS_MAGIC_BIN_BYTES_TO_ALIGN \ + (offsetof(ErtsMagicBinary,u.aligned.data) - \ + offsetof(ErtsMagicBinary,u.unaligned.data)) typedef union { Binary binary; @@ -1306,6 +1321,7 @@ void erts_stale_drv_select(Eterm, ErlDrvPort, ErlDrvEvent, int, int); Port *erts_get_heart_port(void); void erts_emergency_close_ports(void); +void erts_ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon); #if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_COUNT) void erts_lcnt_enable_io_lock_count(int enable); diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index 33b74f30b7..7fa0280396 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -4192,8 +4192,8 @@ static void sweep_one_monitor(ErtsMonitor *mon, void *vpsc) ErtsMonitor *rmon; Process *rp; - ASSERT(is_internal_pid(mon->pid)); - rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK); + ASSERT(is_internal_pid(mon->u.pid)); + rp = erts_pid2proc(NULL, 0, mon->u.pid, ERTS_PROC_LOCK_LINK); if (!rp) { goto done; } @@ -4294,7 +4294,7 @@ port_fire_one_monitor(ErtsMonitor *mon, void *ctx0) Process *origin; ErtsProcLocks origin_locks; - if (mon->type != MON_TARGET || ! is_pid(mon->pid)) { + if (mon->type != MON_TARGET || ! is_pid(mon->u.pid)) { return; } /* @@ -4303,7 +4303,7 @@ port_fire_one_monitor(ErtsMonitor *mon, void *ctx0) */ origin_locks = ERTS_PROC_LOCKS_MSG_SEND | ERTS_PROC_LOCK_LINK; - origin = erts_pid2proc(NULL, 0, mon->pid, origin_locks); + origin = erts_pid2proc(NULL, 0, mon->u.pid, origin_locks); if (origin) { DeclareTmpHeapNoproc(lhp,3); SweepContext *ctx = (SweepContext *)ctx0; @@ -5485,7 +5485,7 @@ typedef struct { static void prt_one_monitor(ErtsMonitor *mon, void *vprtd) { prt_one_lnk_data *prtd = (prt_one_lnk_data *) vprtd; - erts_print(prtd->to, prtd->arg, "(%T,%T)", mon->pid,mon->ref); + erts_print(prtd->to, prtd->arg, "(%T,%T)", mon->u.pid, mon->ref); } static void prt_one_lnk(ErtsLink *lnk, void *vprtd) @@ -7630,7 +7630,7 @@ erl_drv_convert_time_unit(ErlDrvTime val, (int) to); } -static void ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon) +void erts_ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon) { RefThing *refp; ASSERT(is_internal_ref(ref)); @@ -7664,7 +7664,7 @@ static int do_driver_monitor_process(Port *prt, erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, ref, prt->common.id, NIL); erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - ref_to_driver_monitor(ref,monitor); + erts_ref_to_driver_monitor(ref,monitor); return 0; } @@ -7713,7 +7713,7 @@ static int do_driver_demonitor_process(Port *prt, Eterm *buf, return 1; } ASSERT(mon->type == MON_ORIGIN); - to = mon->pid; + to = mon->u.pid; ASSERT(is_internal_pid(to)); rp = erts_pid2proc_opt(NULL, 0, @@ -7775,7 +7775,7 @@ static ErlDrvTermData do_driver_get_monitored_process(Port *prt, Eterm *buf, return driver_term_nil; } ASSERT(mon->type == MON_ORIGIN); - to = mon->pid; + to = mon->u.pid; ASSERT(is_internal_pid(to)); return (ErlDrvTermData) to; } @@ -7831,7 +7831,7 @@ void erts_fire_port_monitor(Port *prt, Eterm ref) } callback = prt->drv_ptr->process_exit; ASSERT(callback != NULL); - ref_to_driver_monitor(ref,&drv_monitor); + erts_ref_to_driver_monitor(ref,&drv_monitor); ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); DRV_MONITOR_UNLOCK_PDL(prt); #ifdef USE_VM_PROBES diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c index 79f567d16c..5f4eca68d0 100644 --- a/erts/emulator/sys/common/erl_check_io.c +++ b/erts/emulator/sys/common/erl_check_io.c @@ -1224,6 +1224,8 @@ ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env, DTRACE_CHARBUF(name, 64); #endif + ASSERT(!resource->dbg_is_dying); + #ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS if ((unsigned)fd >= (unsigned)erts_smp_atomic_read_nob(&drv_ev_state_len)) { if (fd < 0) { diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl index 27276e6646..e3af9f7454 100644 --- a/erts/emulator/test/nif_SUITE.erl +++ b/erts/emulator/test/nif_SUITE.erl @@ -32,6 +32,12 @@ basic/1, reload_error/1, upgrade/1, heap_frag/1, t_on_load/1, select/1, + monitor_process_a/1, + monitor_process_b/1, + monitor_process_c/1, + monitor_process_d/1, + demonitor_process/1, + monitor_frenzy/1, hipe/1, types/1, many_args/1, binaries/1, get_string/1, get_atom/1, maps/1, @@ -57,6 +63,8 @@ -define(nif_stub,nif_stub_error(?LINE)). +-define(is_resource, is_binary). % to be is_reference + suite() -> [{ct_hooks,[ts_install_cth]}]. all() -> @@ -66,6 +74,8 @@ all() -> ++ [reload_error, heap_frag, types, many_args, select, + {group, monitor}, + monitor_frenzy, hipe, binaries, get_string, get_atom, maps, api_macros, from_array, iolist_as_binary, resource, resource_binary, @@ -83,13 +93,19 @@ all() -> nif_snprintf]. groups() -> - [{G, [], api_repeaters()} || G <- api_groups()]. + [{G, [], api_repeaters()} || G <- api_groups()] + ++ + [{monitor, [], [monitor_process_a, + monitor_process_b, + monitor_process_c, + monitor_process_d, + demonitor_process]}]. + api_groups() -> [api_latest, api_2_4, api_2_0]. api_repeaters() -> [upgrade, resource_takeover, t_on_load]. -init_per_group(api_latest, Config) -> Config; init_per_group(api_2_4, Config) -> [{nif_api_version, ".2_4"} | Config]; init_per_group(api_2_0, Config) -> @@ -99,7 +115,8 @@ init_per_group(api_2_0, Config) -> {skip, "API 2.0 buggy on Windows 64-bit"}; _ -> [{nif_api_version, ".2_0"} | Config] - end. + end; +init_per_group(_, Config) -> Config. end_per_group(_,_) -> ok. @@ -559,6 +576,216 @@ write_full(W, C, Acc) -> Acc end. +%% Basic monitoring of one process that terminates +monitor_process_a(Config) -> + ensure_lib_loaded(Config), + + F = fun(Terminator, UseMsgEnv) -> + Pid = spawn(fun() -> + receive + {exit, Arg} -> exit(Arg); + return -> ok; + BadMatch -> goodmatch = BadMatch + end + end), + R_ptr = alloc_monitor_resource_nif(), + {0, Mon} = monitor_process_nif(R_ptr, Pid, UseMsgEnv, self()), + [R_ptr] = monitored_by(Pid), + Terminator(Pid), + [{monitor_resource_down, R_ptr, Pid, Mon}] = flush(), + [] = last_resource_dtor_call(), + ok = release_resource(R_ptr), + {R_ptr, _, 1} = last_resource_dtor_call() + end, + + T1 = fun(Pid) -> Pid ! {exit, 17} end, + T2 = fun(Pid) -> Pid ! return end, + T3 = fun(Pid) -> Pid ! badmatch end, + T4 = fun(Pid) -> exit(Pid, 18) end, + + [F(T, UME) || T <- [T1,T2,T3,T4], UME <- [true, false]], + + ok. + +%% Test auto-demonitoring at resource destruction +monitor_process_b(Config) -> + ensure_lib_loaded(Config), + + Pid = spawn_link(fun() -> + receive + return -> ok + end + end), + R_ptr = alloc_monitor_resource_nif(), + {0,_} = monitor_process_nif(R_ptr, Pid, true, self()), + [R_ptr] = monitored_by(Pid), + ok = release_resource(R_ptr), + [] = flush(), + {R_ptr, _, 1} = last_resource_dtor_call(), + [] = monitored_by(Pid), + Pid ! return, + ok. + +%% Test termination of monitored process holding last resource ref +monitor_process_c(Config) -> + ensure_lib_loaded(Config), + + Papa = self(), + Pid = spawn_link(fun() -> + R_ptr = alloc_monitor_resource_nif(), + {0,Mon} = monitor_process_nif(R_ptr, self(), true, Papa), + [R_ptr] = monitored_by(self()), + put(store, make_resource(R_ptr)), + ok = release_resource(R_ptr), + [] = last_resource_dtor_call(), + Papa ! {self(), done, R_ptr, Mon}, + exit + end), + [{Pid, done, R_ptr, Mon}, + {monitor_resource_down, R_ptr, Pid, Mon}] = flush(), + {R_ptr, _, 1} = last_resource_dtor_call(), + ok. + +%% Test race of resource dtor called when monitored process is exiting +monitor_process_d(Config) -> + ensure_lib_loaded(Config), + + Papa = self(), + {Target,TRef} = spawn_monitor(fun() -> + nothing = receive_any() + end), + + R_ptr = alloc_monitor_resource_nif(), + {0,_} = monitor_process_nif(R_ptr, Target, true, self()), + [Papa, R_ptr] = monitored_by(Target), + + exit(Target, die), + ok = release_resource(R_ptr), + + [{'DOWN', TRef, process, Target, die}] = flush(), %% no monitor_resource_down + {R_ptr, _, 1} = last_resource_dtor_call(), + + ok. + +%% Test basic demonitoring +demonitor_process(Config) -> + ensure_lib_loaded(Config), + + Pid = spawn_link(fun() -> + receive + return -> ok + end + end), + R_ptr = alloc_monitor_resource_nif(), + {0,MonBin1} = monitor_process_nif(R_ptr, Pid, true, self()), + [R_ptr] = monitored_by(Pid), + {0,MonBin2} = monitor_process_nif(R_ptr, Pid, true, self()), + [R_ptr, R_ptr] = monitored_by(Pid), + 0 = demonitor_process_nif(R_ptr, MonBin1), + [R_ptr] = monitored_by(Pid), + 1 = demonitor_process_nif(R_ptr, MonBin1), + 0 = demonitor_process_nif(R_ptr, MonBin2), + [] = monitored_by(Pid), + 1 = demonitor_process_nif(R_ptr, MonBin2), + + ok = release_resource(R_ptr), + [] = flush(), + {R_ptr, _, 1} = last_resource_dtor_call(), + [] = monitored_by(Pid), + Pid ! return, + ok. + + +monitored_by(Pid) -> + {monitored_by, List0} = process_info(Pid, monitored_by), + List1 = lists:map(fun(E) when ?is_resource(E) -> + {Ptr, _} = get_resource(monitor_resource_type, E), + Ptr; + (E) -> E + end, + List0), + erlang:garbage_collect(), + lists:sort(List1). + +-define(FRENZY_RAND_BITS, 25). + +monitor_frenzy(Config) -> + ensure_lib_loaded(Config), + + Procs1 = processes(), + io:format("~p processes before: ~p\n", [length(Procs1), Procs1]), + + %% Spawn first worker process + Master = self(), + spawn_link(fun() -> + SelfPix = monitor_frenzy_nif(init, ?FRENZY_RAND_BITS, 0, 0), + unlink(Master), + frenzy(SelfPix, undefined) + end), + receive after 5*1000 -> ok end, + + io:format("stats = ~p\n", [monitor_frenzy_nif(stats, 0, 0, 0)]), + + Pids = monitor_frenzy_nif(stop, 0, 0, 0), + io:format("stats = ~p\n", [monitor_frenzy_nif(stats, 0, 0, 0)]), + + lists:foreach(fun(P) -> exit(P, stop) end, Pids), + + io:format("stats = ~p\n", [monitor_frenzy_nif(stats, 0, 0, 0)]), + + Procs2 = processes(), + io:format("~p processes after: ~p\n", [length(Procs2), Procs2]), + ok. + + +frenzy(_SelfPix, done) -> + ok; +frenzy(SelfPix, State0) -> + Rnd = rand:uniform(1 bsl (?FRENZY_RAND_BITS+2)) - 1, + Op = Rnd band 3, + State1 = frenzy_do_op(SelfPix, Op, (Rnd bsr 2), State0), + frenzy(SelfPix, State1). + +frenzy_do_op(SelfPix, Op, Rnd, Pid0) -> + case Op of + 0 -> % add/remove process + Papa = self(), + NewPid = case Pid0 of + undefined -> % Prepare new process to be added + spawn(fun() -> + MRef = monitor(process, Papa), + case receive_any() of + {go, MyPix, MyState} -> + demonitor(MRef, [flush]), + frenzy(MyPix, MyState); + {'DOWN', MRef, process, Papa, _} -> + ok + end + end); + _ -> + Pid0 + end, + case monitor_frenzy_nif(Op, Rnd, SelfPix, NewPid) of + NewPix when is_integer(NewPix) -> + NewPid ! {go, NewPix, undefined}, + undefined; + ExitPid when is_pid(ExitPid) -> + false = (ExitPid =:= self()), + exit(ExitPid,die), + NewPid; + done -> + done + end; + _ -> + case monitor_frenzy_nif(Op, Rnd, SelfPix, undefined) of + ok -> Pid0; + 0 -> Pid0; + 1 -> Pid0; + done -> done + end + end. + + hipe(Config) when is_list(Config) -> Data = proplists:get_value(data_dir, Config), Priv = proplists:get_value(priv_dir, Config), @@ -814,6 +1041,7 @@ maps(Config) when is_list(Config) -> {1, M2} = make_map_remove_nif(M2, "key3"), {0, undefined} = make_map_remove_nif(self(), key), + verify_tmpmem(TmpMem), ok. %% Test macros enif_make_list and enif_make_tuple @@ -1302,7 +1530,7 @@ resource_takeover(Config) when is_list(Config) -> [{load,1,1,101},{get_priv_data_ptr,1,2,102}] = nif_mod_call_history(), {NA7,BinNA7} = make_resource(0, Holder, "NA7"), - {AN7,BinAN7} = make_resource(1, Holder, "AN7"), + {AN7,_BinAN7} = make_resource(1, Holder, "AN7"), ok = forget_resource(NA7), [{{resource_dtor_A_v1,BinNA7},1,_,_}] = nif_mod_call_history(), @@ -1793,7 +2021,7 @@ otp_9828(Config) -> ensure_lib_loaded(Config, 1), otp_9828_loop(<<"I'm alive!">>, 1000). -otp_9828_loop(Bin, 0) -> +otp_9828_loop(_Bin, 0) -> ok; otp_9828_loop(Bin, Val) -> WrtBin = <>, @@ -2075,7 +2303,7 @@ nif_raise_exceptions(NifFunc) -> -define(ERL_NIF_TIME_ERROR, -9223372036854775808). -define(TIME_UNITS, [second, millisecond, microsecond, nanosecond]). -nif_monotonic_time(Config) -> +nif_monotonic_time(_Config) -> ?ERL_NIF_TIME_ERROR = monotonic_time(invalid_time_unit), mtime_loop(1000000). @@ -2100,7 +2328,7 @@ chk_mtime([TU|TUs]) -> end, chk_mtime(TUs). -nif_time_offset(Config) -> +nif_time_offset(_Config) -> ?ERL_NIF_TIME_ERROR = time_offset(invalid_time_unit), toffs_loop(1000000). @@ -2138,7 +2366,7 @@ chk_toffs([TU|TUs]) -> end, chk_toffs(TUs). -nif_convert_time_unit(Config) -> +nif_convert_time_unit(_Config) -> ?ERL_NIF_TIME_ERROR = convert_time_unit(0, second, invalid_time_unit), ?ERL_NIF_TIME_ERROR = convert_time_unit(0, invalid_time_unit, second), ?ERL_NIF_TIME_ERROR = convert_time_unit(0, invalid_time_unit, invalid_time_unit), @@ -2413,6 +2641,10 @@ write_nif(_,_) -> ?nif_stub. read_nif(_,_) -> ?nif_stub. is_closed_nif(_) -> ?nif_stub. last_fd_stop_call() -> ?nif_stub. +alloc_monitor_resource_nif() -> ?nif_stub. +monitor_process_nif(_,_,_,_) -> ?nif_stub. +demonitor_process_nif(_,_) -> ?nif_stub. +monitor_frenzy_nif(_,_,_,_) -> ?nif_stub. %% maps is_map_nif(_) -> ?nif_stub. diff --git a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c index c4f9611ec8..ee925512d2 100644 --- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c +++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c @@ -31,6 +31,27 @@ #include "nif_mod.h" +#if 0 +static ErlNifMutex* dbg_trace_lock; +#define DBG_TRACE_INIT dbg_trace_lock = enif_mutex_create("nif_SUITE.DBG_TRACE") +#define DBG_TRACE_FINI enif_mutex_destroy(dbg_trace_lock) +#define DBG_TRACE_LOCK enif_mutex_lock(dbg_trace_lock) +#define DBG_TRACE_UNLOCK enif_mutex_unlock(dbg_trace_lock) +#define DBG_TRACE0(FMT) do {DBG_TRACE_LOCK; enif_fprintf(stderr, FMT); DBG_TRACE_UNLOCK; }while(0) +#define DBG_TRACE1(FMT, A) do {DBG_TRACE_LOCK; enif_fprintf(stderr, FMT, A); DBG_TRACE_UNLOCK; }while(0) +#define DBG_TRACE2(FMT, A, B) do {DBG_TRACE_LOCK; enif_fprintf(stderr, FMT, A, B); DBG_TRACE_UNLOCK; }while(0) +#define DBG_TRACE3(FMT, A, B, C) do {DBG_TRACE_LOCK; enif_fprintf(stderr, FMT, A, B, C); DBG_TRACE_UNLOCK; }while(0) +#define DBG_TRACE4(FMT, A, B, C, D) do {DBG_TRACE_LOCK; enif_fprintf(stderr, FMT, A, B, C, D); DBG_TRACE_UNLOCK; }while(0) +#else +#define DBG_TRACE_INIT +#define DBG_TRACE_FINI +#define DBG_TRACE0(FMT) +#define DBG_TRACE1(FMT, A) +#define DBG_TRACE2(FMT, A, B) +#define DBG_TRACE3(FMT, A, B, C) +#define DBG_TRACE4(FMT, A, B, C, D) +#endif + static int static_cntA; /* zero by default */ static int static_cntB = NIF_SUITE_LIB_VER * 100; @@ -48,6 +69,12 @@ static ERL_NIF_TERM atom_eagain; static ERL_NIF_TERM atom_eof; static ERL_NIF_TERM atom_error; static ERL_NIF_TERM atom_fd_resource_stop; +static ERL_NIF_TERM atom_monitor_resource_type; +static ERL_NIF_TERM atom_monitor_resource_down; +static ERL_NIF_TERM atom_init; +static ERL_NIF_TERM atom_stats; +static ERL_NIF_TERM atom_done; +static ERL_NIF_TERM atom_stop; typedef struct { @@ -120,6 +147,27 @@ struct fd_resource { ErlNifPid pid; }; +static ErlNifResourceType* monitor_resource_type; +static void monitor_resource_dtor(ErlNifEnv* env, void* obj); +static void monitor_resource_down(ErlNifEnv*, void* obj, ErlNifPid*, ErlNifMonitor*); +static ErlNifResourceTypeInit monitor_rt_init = { + monitor_resource_dtor, + NULL, + monitor_resource_down +}; +struct monitor_resource { + ErlNifPid receiver; + int use_msgenv; +}; + +static ErlNifResourceType* frenzy_resource_type; +static void frenzy_resource_dtor(ErlNifEnv* env, void* obj); +static void frenzy_resource_down(ErlNifEnv*, void* obj, ErlNifPid*, ErlNifMonitor*); +static ErlNifResourceTypeInit frenzy_rt_init = { + frenzy_resource_dtor, + NULL, + frenzy_resource_down +}; static int get_pointer(ErlNifEnv* env, ERL_NIF_TERM term, void** pp) { @@ -148,6 +196,8 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) data->call_history = NULL; data->nif_mod = NULL; + DBG_TRACE_INIT; + add_call(env, data, "load"); data->rt_arr[0].t = enif_open_resource_type(env,NULL,"Gold",resource_dtor, @@ -165,6 +215,13 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) fd_resource_type = enif_open_resource_type_x(env, "nif_SUITE.fd", &fd_rt_init, ERL_NIF_RT_CREATE, NULL); + monitor_resource_type = enif_open_resource_type_x(env, "nif_SUITE.monitor", + &monitor_rt_init, + ERL_NIF_RT_CREATE, NULL); + frenzy_resource_type = enif_open_resource_type_x(env, "nif_SUITE.monitor_frenzy", + &frenzy_rt_init, + ERL_NIF_RT_CREATE, NULL); + atom_false = enif_make_atom(env,"false"); atom_true = enif_make_atom(env,"true"); atom_self = enif_make_atom(env,"self"); @@ -179,6 +236,12 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) atom_eof = enif_make_atom(env, "eof"); atom_error = enif_make_atom(env, "error"); atom_fd_resource_stop = enif_make_atom(env, "fd_resource_stop"); + atom_monitor_resource_type = enif_make_atom(env, "monitor_resource_type"); + atom_monitor_resource_down = enif_make_atom(env, "monitor_resource_down"); + atom_init = enif_make_atom(env,"init"); + atom_stats = enif_make_atom(env,"stats"); + atom_done = enif_make_atom(env,"done"); + atom_stop = enif_make_atom(env,"stop"); *priv_data = data; return 0; @@ -232,6 +295,7 @@ static void unload(ErlNifEnv* env, void* priv_data) } enif_free(priv_data); } + DBG_TRACE_FINI; } static ERL_NIF_TERM lib_version(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) @@ -855,6 +919,9 @@ static ERL_NIF_TERM get_resource(ErlNifEnv* env, int argc, const ERL_NIF_TERM ar if (enif_is_identical(argv[0], atom_binary_resource_type)) { type.t = binary_resource_type; } + else if (enif_is_identical(argv[0], atom_monitor_resource_type)) { + type.t = monitor_resource_type; + } else { get_pointer(env, argv[0], &type.vp); } @@ -1496,7 +1563,6 @@ static ERL_NIF_TERM send_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[ static ERL_NIF_TERM send_copy_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) { - ErlNifEnv* menv; ErlNifPid pid; int ret; if (!enif_get_local_pid(env, argv[0], &pid)) { @@ -2236,6 +2302,472 @@ static ERL_NIF_TERM last_fd_stop_call(ErlNifEnv* env, int argc, const ERL_NIF_TE } +static void monitor_resource_dtor(ErlNifEnv* env, void* obj) +{ + resource_dtor(env, obj); +} + +static ERL_NIF_TERM make_monitor(ErlNifEnv* env, const ErlNifMonitor* mon) +{ + ERL_NIF_TERM mon_bin; + memcpy(enif_make_new_binary(env, sizeof(ErlNifMonitor), &mon_bin), + mon, sizeof(ErlNifMonitor)); + return mon_bin; +} + +static int get_monitor(ErlNifEnv* env, ERL_NIF_TERM term, ErlNifMonitor* mon) +{ + ErlNifBinary bin; + if (!enif_inspect_binary(env, term, &bin) + || bin.size != sizeof(ErlNifMonitor)) + return 0; + memcpy(mon, bin.data, bin.size); + return 1; +} + +static void monitor_resource_down(ErlNifEnv* env, void* obj, ErlNifPid* pid, + ErlNifMonitor* mon) +{ + struct monitor_resource* rsrc = (struct monitor_resource*)obj; + ErlNifEnv* build_env; + ErlNifEnv* msg_env; + ERL_NIF_TERM msg; + + if (rsrc->use_msgenv) { + msg_env = enif_alloc_env(); + build_env = msg_env; + } + else { + msg_env = NULL; + build_env = env; + } + + msg = enif_make_tuple4(build_env, + atom_monitor_resource_down, + make_pointer(build_env, obj), + enif_make_pid(build_env, pid), + make_monitor(build_env, mon)); + + enif_send(env, &rsrc->receiver, msg_env, msg); + if (msg_env) + enif_free_env(msg_env); +} + +static ERL_NIF_TERM alloc_monitor_resource_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) +{ + struct monitor_resource* rsrc; + + rsrc = enif_alloc_resource(monitor_resource_type, sizeof(struct monitor_resource)); + + return make_pointer(env,rsrc); +} + +static ERL_NIF_TERM monitor_process_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) +{ + struct monitor_resource* rsrc; + ErlNifPid target; + ErlNifMonitor mon; + int res; + + if (!get_pointer(env, argv[0], (void**)&rsrc) + || !enif_get_local_pid(env, argv[1], &target) + || !enif_get_local_pid(env, argv[3], &rsrc->receiver)) { + return enif_make_badarg(env); + } + + rsrc->use_msgenv = (argv[2] == atom_true); + res = enif_monitor_process(env, rsrc, &target, &mon); + + return enif_make_tuple2(env, enif_make_int(env, res), make_monitor(env, &mon)); +} + +static ERL_NIF_TERM demonitor_process_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) +{ + struct monitor_resource* rsrc; + ErlNifMonitor mon; + int res; + + if (!get_pointer(env, argv[0], (void**)&rsrc) + || !get_monitor(env, argv[1], &mon)) { + return enif_make_badarg(env); + } + + res = enif_demonitor_process(env, rsrc, &mon); + + return enif_make_int(env, res); +} + +/*********** monitor_frenzy ************/ + +struct frenzy_rand_bits +{ + unsigned int source; + unsigned int bits_consumed; +}; + +static unsigned int frenzy_rand_bits_max; + +unsigned rand_bits(struct frenzy_rand_bits* rnd, unsigned int nbits) +{ + unsigned int res; + + rnd->bits_consumed += nbits; + assert(rnd->bits_consumed <= frenzy_rand_bits_max); + res = rnd->source & ((1 << nbits)-1); + rnd->source >>= nbits; + return res; +} + +#define FRENZY_PROCS_MAX_BITS 4 +#define FRENZY_PROCS_MAX (1 << FRENZY_PROCS_MAX_BITS) + +#define FRENZY_RESOURCES_MAX_BITS 4 +#define FRENZY_RESOURCES_MAX (1 << FRENZY_RESOURCES_MAX_BITS) + +#define FRENZY_MONITORS_MAX_BITS 4 +#define FRENZY_MONITORS_MAX (1 << FRENZY_MONITORS_MAX_BITS) + +struct frenzy_monitor { + ErlNifMutex* lock; + enum { + MON_FREE, MON_FREE_DOWN, MON_FREE_DEMONITOR, + MON_TRYING, MON_ACTIVE, MON_PENDING + } state; + ErlNifMonitor mon; + ErlNifPid pid; + unsigned int use_cnt; +}; + +struct frenzy_resource { + unsigned int rix; + struct frenzy_monitor monv[FRENZY_MONITORS_MAX]; +}; +struct frenzy_reslot { + ErlNifMutex* lock; + struct frenzy_resource* obj; + unsigned long alloc_cnt; + unsigned long release_cnt; +}; +static struct frenzy_reslot resv[FRENZY_RESOURCES_MAX]; + +static ERL_NIF_TERM monitor_frenzy_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) +{ + struct frenzy_proc { + ErlNifPid pid; + int is_free; + }; + static struct frenzy_proc procs[FRENZY_PROCS_MAX]; + static struct frenzy_proc* proc_refs[FRENZY_PROCS_MAX]; + static unsigned int nprocs, old_nprocs; + static ErlNifMutex* procs_lock; + static unsigned long spawn_cnt = 0; + static unsigned long kill_cnt = 0; + static unsigned long proc_histogram[FRENZY_PROCS_MAX]; + + static const unsigned int primes[] = {7, 13, 17, 19}; + + struct frenzy_resource* r; + struct frenzy_rand_bits rnd; + unsigned int op, inc, my_nprocs; + unsigned int mix; /* r->monv[] index */ + unsigned int rix; /* resv[] index */ + unsigned int pix; /* procs[] index */ + unsigned int ref_ix; /* proc_refs[] index */ + int self_pix, rv; + ERL_NIF_TERM retval = atom_error; + const ERL_NIF_TERM Op = argv[0]; + const ERL_NIF_TERM Rnd = argv[1]; + const ERL_NIF_TERM SelfPix = argv[2]; + const ERL_NIF_TERM NewPid = argv[3]; + + if (enif_is_atom(env, Op)) { + if (Op == atom_init) { + if (procs_lock || !enif_get_uint(env, Rnd, &frenzy_rand_bits_max)) + return enif_make_badarg(env); + + procs_lock = enif_mutex_create("nif_SUITE:monitor_frenzy.procs"); + nprocs = 0; + old_nprocs = 0; + for (pix = 0; pix < FRENZY_PROCS_MAX; pix++) { + proc_refs[pix] = &procs[pix]; + procs[pix].is_free = 1; + proc_histogram[pix] = 0; + } + for (rix = 0; rix < FRENZY_RESOURCES_MAX; rix++) { + resv[rix].lock = enif_mutex_create("nif_SUITE:monitor_frenzy.resv.lock"); + resv[rix].obj = NULL; + resv[rix].alloc_cnt = 0; + resv[rix].release_cnt = 0; + } + + /* Add self as first process */ + enif_self(env, &procs[0].pid); + procs[0].is_free = 0; + old_nprocs = ++nprocs; + + spawn_cnt = 1; + kill_cnt = 0; + return enif_make_uint(env, 0); /* SelfPix */ + } + else if (Op == atom_stats) { + ERL_NIF_TERM hist[FRENZY_PROCS_MAX]; + unsigned long res_alloc_cnt = 0; + unsigned long res_release_cnt = 0; + for (ref_ix = 0; ref_ix < FRENZY_PROCS_MAX; ref_ix++) { + hist[ref_ix] = enif_make_ulong(env, proc_histogram[ref_ix]); + } + for (rix = 0; rix < FRENZY_RESOURCES_MAX; rix++) { + res_alloc_cnt += resv[rix].alloc_cnt; + res_release_cnt += resv[rix].release_cnt; + } + + return + enif_make_list4(env, + enif_make_tuple2(env, enif_make_string(env, "proc_histogram", ERL_NIF_LATIN1), + enif_make_list_from_array(env, hist, FRENZY_PROCS_MAX)), + enif_make_tuple2(env, enif_make_string(env, "spawn_cnt", ERL_NIF_LATIN1), + enif_make_ulong(env, spawn_cnt)), + enif_make_tuple2(env, enif_make_string(env, "kill_cnt", ERL_NIF_LATIN1), + enif_make_ulong(env, kill_cnt)), + enif_make_tuple3(env, enif_make_string(env, "resource_alloc", ERL_NIF_LATIN1), + enif_make_ulong(env, res_alloc_cnt), + enif_make_ulong(env, res_release_cnt))); + + } + else if (Op == atom_stop && procs_lock) { /* stop all */ + retval = enif_make_list(env, 0); + enif_mutex_lock(procs_lock); + for (ref_ix = 0; ref_ix < nprocs; ref_ix++) { + assert(!proc_refs[ref_ix]->is_free); + retval = enif_make_list_cell(env, enif_make_pid(env, &proc_refs[ref_ix]->pid), + retval); + proc_refs[ref_ix]->is_free = 1; + } + kill_cnt += nprocs; + nprocs = 0; + old_nprocs = 0; + enif_mutex_unlock(procs_lock); + return retval; + } + return enif_make_badarg(env); + } + + if (!enif_get_int(env, SelfPix, &self_pix) || + !enif_get_uint(env, Op, &op) || + !enif_get_uint(env, Rnd, &rnd.source)) + return enif_make_badarg(env); + + rnd.bits_consumed = 0; + switch (op) { + case 0: { /* add/remove process */ + ErlNifPid self; + enif_self(env, &self); + + ref_ix = rand_bits(&rnd, FRENZY_PROCS_MAX_BITS) % FRENZY_PROCS_MAX; + enif_mutex_lock(procs_lock); + if (procs[self_pix].is_free || procs[self_pix].pid.pid != self.pid) { + /* Some one already removed me */ + enif_mutex_unlock(procs_lock); + return atom_done; + } + if (ref_ix >= nprocs || nprocs < 2) { /* add process */ + ref_ix = nprocs++; + pix = proc_refs[ref_ix] - procs; + assert(procs[pix].is_free); + if (!enif_get_local_pid(env, NewPid, &procs[pix].pid)) + abort(); + procs[pix].is_free = 0; + spawn_cnt++; + proc_histogram[ref_ix]++; + old_nprocs = nprocs; + enif_mutex_unlock(procs_lock); + DBG_TRACE2("Add pid %T, nprocs = %u\n", NewPid, nprocs); + retval = enif_make_uint(env, pix); + } + else { /* remove process */ + pix = proc_refs[ref_ix] - procs; + if (pix == self_pix) { + ref_ix = (ref_ix + 1) % nprocs; + pix = proc_refs[ref_ix] - procs; + } + assert(procs[pix].pid.pid != self.pid); + assert(!procs[pix].is_free); + retval = enif_make_pid(env, &procs[pix].pid); + --nprocs; + assert(!proc_refs[nprocs]->is_free); + if (ref_ix != nprocs) { + struct frenzy_proc* tmp = proc_refs[ref_ix]; + proc_refs[ref_ix] = proc_refs[nprocs]; + proc_refs[nprocs] = tmp; + } + procs[pix].is_free = 1; + proc_histogram[nprocs]++; + kill_cnt++; + enif_mutex_unlock(procs_lock); + DBG_TRACE2("Removed pid %T, nprocs = %u\n", retval, nprocs); + } + break; + } + case 1: + case 2: /* create/delete/lookup resource */ + rix = rand_bits(&rnd, FRENZY_RESOURCES_MAX_BITS) % FRENZY_RESOURCES_MAX; + inc = primes[rand_bits(&rnd, 2)]; + while (enif_mutex_trylock(resv[rix].lock) == EBUSY) { + rix = (rix + inc) % FRENZY_RESOURCES_MAX; + } + if (resv[rix].obj == NULL) { + r = enif_alloc_resource(frenzy_resource_type, + sizeof(struct frenzy_resource)); + resv[rix].obj = r; + resv[rix].alloc_cnt++; + r->rix = rix; + for (mix = 0; mix < FRENZY_MONITORS_MAX; mix++) { + r->monv[mix].lock = enif_mutex_create("nif_SUITE:monitor_frenzy.monv.lock"); + r->monv[mix].state = MON_FREE; + r->monv[mix].use_cnt = 0; + r->monv[mix].pid.pid = 0; /* null-pid */ + } + DBG_TRACE2("New resource at r=%p rix=%u\n", r, rix); + } + else if (rand_bits(&rnd, 3) == 0) { + r = resv[rix].obj; + resv[rix].obj = NULL; + resv[rix].release_cnt++; + enif_mutex_unlock(resv[rix].lock); + DBG_TRACE2("Delete resource at r=%p rix=%u\n", r, rix); + enif_release_resource(r); + retval = atom_ok; + break; + } + else { + r = resv[rix].obj; + } + enif_keep_resource(r); + enif_mutex_unlock(resv[rix].lock); + + /* monitor/demonitor */ + + mix = rand_bits(&rnd, FRENZY_MONITORS_MAX_BITS) % FRENZY_MONITORS_MAX; + inc = primes[rand_bits(&rnd, 2)]; + while (enif_mutex_trylock(r->monv[mix].lock) == EBUSY) { + mix = (mix + inc) % FRENZY_MONITORS_MAX; + } + switch (r->monv[mix].state) { + case MON_FREE: + case MON_FREE_DOWN: + case MON_FREE_DEMONITOR: { /* do monitor */ + /* + * Use an old possibly larger value of 'nprocs', to increase + * probability of monitoring an already terminated process + */ + my_nprocs = old_nprocs; + if (my_nprocs > 0) { + int save_state = r->monv[mix].state; + ref_ix = rand_bits(&rnd, FRENZY_PROCS_MAX_BITS) % my_nprocs; + pix = proc_refs[ref_ix] - procs; + r->monv[mix].pid.pid = procs[pix].pid.pid; /* "atomic" */ + r->monv[mix].state = MON_TRYING; + rv = enif_monitor_process(env, r, &r->monv[mix].pid, &r->monv[mix].mon); + if (rv == 0) { + r->monv[mix].state = MON_ACTIVE; + r->monv[mix].use_cnt++; + DBG_TRACE3("Monitor from r=%p rix=%u to %T\n", + r, r->rix, r->monv[mix].pid.pid); + } + else { + r->monv[mix].state = save_state; + DBG_TRACE4("Monitor from r=%p rix=%u to %T FAILED with %d\n", + r, r->rix, r->monv[mix].pid.pid, rv); + } + retval = enif_make_int(env,rv); + } + else { + DBG_TRACE0("No pids to monitor\n"); + retval = atom_ok; + } + break; + } + case MON_ACTIVE: /* do demonitor */ + rv = enif_demonitor_process(env, r, &r->monv[mix].mon); + if (rv == 0) { + DBG_TRACE3("Demonitor from r=%p rix=%u to %T\n", + r, r->rix, r->monv[mix].pid.pid); + r->monv[mix].state = MON_FREE_DEMONITOR; + } + else { + DBG_TRACE4("Demonitor from r=%p rix=%u to %T FAILED with %d\n", + r, r->rix, r->monv[mix].pid.pid, rv); + r->monv[mix].state = MON_PENDING; + } + retval = enif_make_int(env,rv); + break; + + case MON_PENDING: /* waiting for 'down' callback, do nothing */ + retval = atom_ok; + break; + default: + abort(); + break; + } + enif_mutex_unlock(r->monv[mix].lock); + enif_release_resource(r); + break; + + case 3: /* no-op */ + retval = atom_ok; + break; + } + + { + int percent = (rand_bits(&rnd, 6) + 1) * 2; /* 2 to 128 */ + if (percent <= 100) + enif_consume_timeslice(env, percent); + } + + return retval; +} + +static void frenzy_resource_dtor(ErlNifEnv* env, void* obj) +{ + struct frenzy_resource* r = (struct frenzy_resource*) obj; + unsigned int mix; + + DBG_TRACE2("DTOR r=%p rix=%u\n", r, r->rix); + + for (mix = 0; mix < FRENZY_MONITORS_MAX; mix++) { + assert(r->monv[mix].state != MON_PENDING); + enif_mutex_destroy(r->monv[mix].lock); + r->monv[mix].lock = NULL; + } + +} + +static void frenzy_resource_down(ErlNifEnv* env, void* obj, ErlNifPid* pid, + ErlNifMonitor* mon) +{ + struct frenzy_resource* r = (struct frenzy_resource*) obj; + unsigned int mix; + + DBG_TRACE3("DOWN pid=%T, r=%p rix=%u\n", pid->pid, r, r->rix); + + for (mix = 0; mix < FRENZY_MONITORS_MAX; mix++) { + if (r->monv[mix].pid.pid == pid->pid && r->monv[mix].state >= MON_TRYING) { + enif_mutex_lock(r->monv[mix].lock); + if (memcmp(mon, &r->monv[mix].mon, sizeof(*mon)) == 0) { + assert(r->monv[mix].state >= MON_ACTIVE); + r->monv[mix].state = MON_FREE_DOWN; + enif_mutex_unlock(r->monv[mix].lock); + return; + } + enif_mutex_unlock(r->monv[mix].lock); + } + } + enif_fprintf(stderr, "DOWN called for unknown monitor\n"); + abort(); +} + + + static ErlNifFunc nif_funcs[] = { {"lib_version", 0, lib_version}, @@ -2318,7 +2850,11 @@ static ErlNifFunc nif_funcs[] = {"write_nif", 2, write_nif}, {"read_nif", 2, read_nif}, {"is_closed_nif", 1, is_closed_nif}, - {"last_fd_stop_call", 0, last_fd_stop_call} + {"last_fd_stop_call", 0, last_fd_stop_call}, + {"alloc_monitor_resource_nif", 0, alloc_monitor_resource_nif}, + {"monitor_process_nif", 4, monitor_process_nif}, + {"demonitor_process_nif", 2, demonitor_process_nif}, + {"monitor_frenzy_nif", 4, monitor_frenzy_nif} }; ERL_NIF_INIT(nif_SUITE,nif_funcs,load,NULL,upgrade,unload) diff --git a/erts/emulator/test/nif_SUITE_data/nif_mod.c b/erts/emulator/test/nif_SUITE_data/nif_mod.c index e6106b6036..04699d3327 100644 --- a/erts/emulator/test/nif_SUITE_data/nif_mod.c +++ b/erts/emulator/test/nif_SUITE_data/nif_mod.c @@ -176,6 +176,7 @@ static void do_load_info(ErlNifEnv* env, ERL_NIF_TERM load_info, int* retvalp) CHECK(enif_is_empty_list(env, head)); } +#if NIF_LIB_VER != 3 static int load(ErlNifEnv* env, void** priv, ERL_NIF_TERM load_info) { NifModPrivData* data; @@ -230,6 +231,7 @@ static void unload(ErlNifEnv* env, void* priv) add_call(env, data, "unload"); NifModPrivData_release(data); } +#endif /* NIF_LIB_VER != 3 */ static ERL_NIF_TERM lib_version(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) { -- cgit v1.2.3