aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam/dist.c
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam/dist.c')
-rw-r--r--erts/emulator/beam/dist.c2013
1 files changed, 1388 insertions, 625 deletions
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 9fddac8980..8593ad867a 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -103,25 +103,16 @@ dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
-#define PASS_THROUGH 'p' /* This code should go */
-
int erts_is_alive; /* System must be blocked on change */
int erts_dist_buf_busy_limit;
/* distribution trap functions */
-Export* dsend2_trap = NULL;
-Export* dsend3_trap = NULL;
-/*Export* dsend_nosuspend_trap = NULL;*/
-Export* dlink_trap = NULL;
-Export* dunlink_trap = NULL;
Export* dmonitor_node_trap = NULL;
-Export* dgroup_leader_trap = NULL;
-Export* dexit_trap = NULL;
Export* dmonitor_p_trap = NULL;
/* local variables */
-
+static Export *dist_ctrl_put_data_trap;
/* forward declarations */
@@ -129,9 +120,10 @@ static void clear_dist_entry(DistEntry*);
static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy);
static void send_nodes_mon_msgs(Process *, Eterm, Eterm, Eterm, Eterm);
static void init_nodes_monitors(void);
+static Sint abort_connection(DistEntry* dep, Uint32 conn_id);
-static erts_smp_atomic_t no_caches;
-static erts_smp_atomic_t no_nodes;
+static erts_atomic_t no_caches;
+static erts_atomic_t no_nodes;
struct {
Eterm reason;
@@ -144,8 +136,8 @@ delete_cache(ErtsAtomCache *cache)
{
if (cache) {
erts_free(ERTS_ALC_T_DCACHE, (void *) cache);
- ASSERT(erts_smp_atomic_read_nob(&no_caches) > 0);
- erts_smp_atomic_dec_nob(&no_caches);
+ ASSERT(erts_atomic_read_nob(&no_caches) > 0);
+ erts_atomic_dec_nob(&no_caches);
}
}
@@ -156,14 +148,12 @@ create_cache(DistEntry *dep)
int i;
ErtsAtomCache *cp;
- ERTS_SMP_LC_ASSERT(
- is_internal_port(dep->cid)
- && erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid)));
+ ERTS_LC_ASSERT(is_nil(dep->cid));
ASSERT(!dep->cache);
dep->cache = cp = (ErtsAtomCache*) erts_alloc(ERTS_ALC_T_DCACHE,
sizeof(ErtsAtomCache));
- erts_smp_atomic_inc_nob(&no_caches);
+ erts_atomic_inc_nob(&no_caches);
for (i = 0; i < sizeof(cp->in_arr)/sizeof(cp->in_arr[0]); i++) {
cp->in_arr[i] = THE_NON_VALUE;
cp->out_arr[i] = THE_NON_VALUE;
@@ -172,15 +162,17 @@ create_cache(DistEntry *dep)
Uint erts_dist_cache_size(void)
{
- return (Uint) erts_smp_atomic_read_mb(&no_caches)*sizeof(ErtsAtomCache);
+ return (Uint) erts_atomic_read_mb(&no_caches)*sizeof(ErtsAtomCache);
}
static ErtsProcList *
-get_suspended_on_de(DistEntry *dep, Uint32 unset_qflgs)
+get_suspended_on_de(DistEntry *dep, erts_aint32_t unset_qflgs)
{
- ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&dep->qlock));
- dep->qflgs &= ~unset_qflgs;
- if (dep->qflgs & ERTS_DE_QFLG_EXIT) {
+ erts_aint32_t qflgs;
+ ERTS_LC_ASSERT(erts_lc_mtx_is_locked(&dep->qlock));
+ qflgs = erts_atomic32_read_band_acqb(&dep->qflgs, ~unset_qflgs);
+ qflgs &= ~unset_qflgs;
+ if (qflgs & ERTS_DE_QFLG_EXIT) {
/* No resume when exit has been scheduled */
return NULL;
}
@@ -283,17 +275,15 @@ static void doit_monitor_net_exits(ErtsMonitor *mon, void *vnecp)
watched = (is_atom(rmon->name)
? TUPLE2(lhp, rmon->name, dep->sysname)
: rmon->u.pid);
-#ifdef ERTS_SMP
rp_locks |= ERTS_PROC_LOCKS_MSG_SEND;
- erts_smp_proc_lock(rp, ERTS_PROC_LOCKS_MSG_SEND);
-#endif
+ erts_proc_lock(rp, ERTS_PROC_LOCKS_MSG_SEND);
erts_queue_monitor_message(rp, &rp_locks, mon->ref, am_process,
watched, am_noconnection);
erts_destroy_monitor(rmon);
}
UnUseTmpHeapNoproc(3);
}
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
done:
erts_destroy_monitor(mon);
}
@@ -342,7 +332,7 @@ static void doit_link_net_exits_sub(ErtsLink *sublnk, void *vlnecp)
trace_proc(NULL, 0, rp, am_getting_unlinked, sublnk->pid);
}
}
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
done:
erts_destroy_link(sublnk);
@@ -384,7 +374,7 @@ static void doit_node_link_net_exits(ErtsLink *lnk, void *vnecp)
rp = erts_proc_lookup(lnk->pid);
if (!rp)
goto done;
- erts_smp_proc_lock(rp, rp_locks);
+ erts_proc_lock(rp, rp_locks);
if (!ERTS_PROC_IS_EXITING(rp)) {
rlnk = erts_remove_link(&ERTS_P_LINKS(rp), name);
if (rlnk != NULL) {
@@ -403,7 +393,7 @@ static void doit_node_link_net_exits(ErtsLink *lnk, void *vnecp)
erts_queue_message(rp, rp_locks, msgp, tup, am_system);
}
}
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
done:
erts_destroy_link(lnk);
@@ -415,16 +405,16 @@ set_node_not_alive(void *unused)
ErlHeapFragment *bp;
Eterm nodename = erts_this_dist_entry->sysname;
- ASSERT(erts_smp_atomic_read_nob(&no_nodes) == 0);
+ ASSERT(erts_atomic_read_nob(&no_nodes) == 0);
- erts_smp_thr_progress_block();
+ erts_thr_progress_block();
erts_set_this_node(am_Noname, 0);
erts_is_alive = 0;
send_nodes_mon_msgs(NULL, am_nodedown, nodename, am_visible, nodedown.reason);
nodedown.reason = NIL;
bp = nodedown.bp;
nodedown.bp = NULL;
- erts_smp_thr_progress_unblock();
+ erts_thr_progress_unblock();
if (bp)
free_message_buffer(bp);
}
@@ -432,7 +422,7 @@ set_node_not_alive(void *unused)
static ERTS_INLINE void
dec_no_nodes(void)
{
- erts_aint_t no = erts_smp_atomic_dec_read_mb(&no_nodes);
+ erts_aint_t no = erts_atomic_dec_read_mb(&no_nodes);
ASSERT(no >= 0);
ASSERT(erts_get_scheduler_id()); /* Need to be a scheduler */
if (no == 0)
@@ -445,12 +435,40 @@ static ERTS_INLINE void
inc_no_nodes(void)
{
#ifdef DEBUG
- erts_aint_t no = erts_smp_atomic_read_nob(&no_nodes);
+ erts_aint_t no = erts_atomic_read_nob(&no_nodes);
ASSERT(erts_is_alive ? no > 0 : no == 0);
#endif
- erts_smp_atomic_inc_mb(&no_nodes);
+ erts_atomic_inc_mb(&no_nodes);
}
-
+
+static void
+kill_dist_ctrl_proc(void *vpid)
+{
+ Eterm pid = (Eterm) vpid;
+ ErtsProcLocks rp_locks = ERTS_PROC_LOCKS_XSIG_SEND;
+ Process *rp = erts_pid2proc(NULL, 0, pid, rp_locks);
+ if (rp) {
+ erts_send_exit_signal(NULL, rp->common.id, rp, &rp_locks,
+ am_kill, NIL, NULL, 0);
+ if (rp_locks)
+ erts_proc_unlock(rp, rp_locks);
+ }
+}
+
+static void
+schedule_kill_dist_ctrl_proc(Eterm pid)
+{
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
+ int sched_id = 1;
+ if (!esdp || ERTS_SCHEDULER_IS_DIRTY(esdp))
+ sched_id = 1;
+ else
+ sched_id = (int) esdp->no;
+ erts_schedule_misc_aux_work(sched_id,
+ kill_dist_ctrl_proc,
+ (void *) (UWord) pid);
+}
+
/*
* proc is currently running or exiting process.
*/
@@ -460,58 +478,83 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
if (dep == erts_this_dist_entry) { /* Net kernel has died (clean up!!) */
DistEntry *tdep;
- int no_dist_port = 0;
+ int no_dist_ctrl;
+ int no_pending;
Eterm nd_reason = (reason == am_no_network
? am_no_network
: am_net_kernel_terminated);
- erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
+ int i = 0;
+ Eterm *dist_ctrl;
+ DistEntry** pending;
+
+ ERTS_UNDEF(dist_ctrl, NULL);
+ ERTS_UNDEF(pending, NULL);
+
+ erts_rwmtx_rlock(&erts_dist_table_rwmtx);
- for (tdep = erts_hidden_dist_entries; tdep; tdep = tdep->next)
- no_dist_port++;
- for (tdep = erts_visible_dist_entries; tdep; tdep = tdep->next)
- no_dist_port++;
+ no_dist_ctrl = (erts_no_of_hidden_dist_entries +
+ erts_no_of_visible_dist_entries);
+ no_pending = erts_no_of_pending_dist_entries;
/* KILL all port controllers */
- if (no_dist_port == 0)
- erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
- else {
- Eterm def_buf[128];
- int i = 0;
- Eterm *dist_port;
-
- if (no_dist_port <= sizeof(def_buf)/sizeof(def_buf[0]))
- dist_port = &def_buf[0];
- else
- dist_port = erts_alloc(ERTS_ALC_T_TMP,
- sizeof(Eterm)*no_dist_port);
+ if (no_dist_ctrl) {
+ dist_ctrl = erts_alloc(ERTS_ALC_T_TMP,
+ sizeof(Eterm)*no_dist_ctrl);
for (tdep = erts_hidden_dist_entries; tdep; tdep = tdep->next) {
- ASSERT(is_internal_port(tdep->cid));
- dist_port[i++] = tdep->cid;
+ ASSERT(is_internal_port(tdep->cid) || is_internal_pid(tdep->cid));
+ ASSERT(i < no_dist_ctrl);
+ dist_ctrl[i++] = tdep->cid;
}
for (tdep = erts_visible_dist_entries; tdep; tdep = tdep->next) {
- ASSERT(is_internal_port(tdep->cid));
- dist_port[i++] = tdep->cid;
+ ASSERT(is_internal_port(tdep->cid) || is_internal_pid(tdep->cid));
+ ASSERT(i < no_dist_ctrl);
+ dist_ctrl[i++] = tdep->cid;
}
- erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
-
- for (i = 0; i < no_dist_port; i++) {
- Port *prt = erts_port_lookup(dist_port[i],
- ERTS_PORT_SFLGS_INVALID_LOOKUP);
- if (!prt)
- continue;
- ASSERT(erts_atomic32_read_nob(&prt->state)
- & ERTS_PORT_SFLG_DISTRIBUTION);
+ ASSERT(i == no_dist_ctrl);
+ }
+ if (no_pending) {
+ pending = erts_alloc(ERTS_ALC_T_TMP, sizeof(DistEntry*)*no_pending);
+ i = 0;
+ for (tdep = erts_pending_dist_entries; tdep; tdep = tdep->next) {
+ ASSERT(is_nil(tdep->cid));
+ ASSERT(i < no_pending);
+ pending[i++] = tdep;
+ erts_ref_dist_entry(tdep);
+ }
+ ASSERT(i == no_pending);
+ }
+ erts_rwmtx_runlock(&erts_dist_table_rwmtx);
+
+ if (no_dist_ctrl) {
+ for (i = 0; i < no_dist_ctrl; i++) {
+ if (is_internal_pid(dist_ctrl[i]))
+ schedule_kill_dist_ctrl_proc(dist_ctrl[i]);
+ else {
+ Port *prt = erts_port_lookup(dist_ctrl[i],
+ ERTS_PORT_SFLGS_INVALID_LOOKUP);
+ if (prt) {
+ ASSERT(erts_atomic32_read_nob(&prt->state)
+ & ERTS_PORT_SFLG_DISTRIBUTION);
+
+ erts_port_exit(NULL, ERTS_PORT_SIG_FLG_FORCE_SCHED,
+ prt, dist_ctrl[i], nd_reason, NULL);
+ }
+ }
+ }
+ erts_free(ERTS_ALC_T_TMP, dist_ctrl);
+ }
- erts_port_exit(NULL, ERTS_PORT_SIG_FLG_FORCE_SCHED,
- prt, dist_port[i], nd_reason, NULL);
- }
+ if (no_pending) {
+ for (i = 0; i < no_pending; i++) {
+ abort_connection(pending[i], pending[i]->connection_id);
+ erts_deref_dist_entry(pending[i]);
+ }
+ erts_free(ERTS_ALC_T_TMP, pending);
+ }
- if (dist_port != &def_buf[0])
- erts_free(ERTS_ALC_T_TMP, dist_port);
- }
/*
- * When last dist port exits, node will be taken
+ * When last dist ctrl exits, node will be taken
* from alive to not alive.
*/
ASSERT(is_nil(nodedown.reason) && !nodedown.bp);
@@ -528,52 +571,51 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
&nodedown.bp->off_heap);
}
}
- else { /* Call from distribution port */
+ else { /* Call from distribution controller (port/process) */
NetExitsContext nec = {dep};
ErtsLink *nlinks;
ErtsLink *node_links;
ErtsMonitor *monitors;
Uint32 flags;
- erts_smp_atomic_set_mb(&dep->dist_cmd_scheduled, 1);
- erts_smp_de_rwlock(dep);
+ erts_atomic_set_mb(&dep->dist_cmd_scheduled, 1);
+ erts_de_rwlock(dep);
- ERTS_SMP_LC_ASSERT(is_internal_port(dep->cid)
- && erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid)));
+ if (is_internal_port(dep->cid)) {
+ ERTS_LC_ASSERT(erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid)));
- if (erts_port_task_is_scheduled(&dep->dist_cmd))
- erts_port_task_abort(&dep->dist_cmd);
+ if (erts_port_task_is_scheduled(&dep->dist_cmd))
+ erts_port_task_abort(&dep->dist_cmd);
+ }
if (dep->status & ERTS_DE_SFLG_EXITING) {
#ifdef DEBUG
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qflgs & ERTS_DE_QFLG_EXIT);
- erts_smp_mtx_unlock(&dep->qlock);
+ ASSERT(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT);
#endif
}
else {
dep->status |= ERTS_DE_SFLG_EXITING;
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(!(dep->qflgs & ERTS_DE_QFLG_EXIT));
- dep->qflgs |= ERTS_DE_QFLG_EXIT;
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_lock(&dep->qlock);
+ ASSERT(!(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT));
+ erts_atomic32_read_bor_relb(&dep->qflgs, ERTS_DE_QFLG_EXIT);
+ erts_mtx_unlock(&dep->qlock);
}
- erts_smp_de_links_lock(dep);
+ erts_de_links_lock(dep);
monitors = dep->monitors;
nlinks = dep->nlinks;
node_links = dep->node_links;
dep->monitors = NULL;
dep->nlinks = NULL;
dep->node_links = NULL;
- erts_smp_de_links_unlock(dep);
+ erts_de_links_unlock(dep);
nodename = dep->sysname;
flags = dep->flags;
erts_set_dist_entry_not_connected(dep);
- erts_smp_de_rwunlock(dep);
+ erts_de_rwunlock(dep);
erts_sweep_monitors(monitors, &doit_monitor_net_exits, (void *) &nec);
erts_sweep_links(nlinks, &doit_link_net_exits, (void *) &nec);
@@ -586,7 +628,6 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
reason == am_normal ? am_connection_closed : reason);
clear_dist_entry(dep);
-
}
dec_no_nodes();
@@ -607,19 +648,15 @@ void init_dist(void)
nodedown.reason = NIL;
nodedown.bp = NULL;
- erts_smp_atomic_init_nob(&no_nodes, 0);
- erts_smp_atomic_init_nob(&no_caches, 0);
+ erts_atomic_init_nob(&no_nodes, 0);
+ erts_atomic_init_nob(&no_caches, 0);
/* Lookup/Install all references to trap functions */
- dsend2_trap = trap_function(am_dsend,2);
- dsend3_trap = trap_function(am_dsend,3);
- /* dsend_nosuspend_trap = trap_function(am_dsend_nosuspend,2);*/
- dlink_trap = trap_function(am_dlink,1);
- dunlink_trap = trap_function(am_dunlink,1);
dmonitor_node_trap = trap_function(am_dmonitor_node,3);
- dgroup_leader_trap = trap_function(am_dgroup_leader,2);
- dexit_trap = trap_function(am_dexit, 2);
dmonitor_p_trap = trap_function(am_dmonitor_p, 2);
+ dist_ctrl_put_data_trap = erts_export_put(am_erts_internal,
+ am_dist_ctrl_put_data,
+ 2);
}
#define ErtsDistOutputBuf2Binary(OB) \
@@ -634,6 +671,7 @@ alloc_dist_obuf(Uint size)
obuf = (ErtsDistOutputBuf *) &bin->orig_bytes[0];
#ifdef DEBUG
obuf->dbg_pattern = ERTS_DIST_OUTPUT_BUF_DBG_PATTERN;
+ obuf->alloc_endp = obuf->data + size;
ASSERT(bin == ErtsDistOutputBuf2Binary(obuf));
#endif
return obuf;
@@ -654,26 +692,11 @@ size_obuf(ErtsDistOutputBuf *obuf)
return bin->orig_size;
}
-static void clear_dist_entry(DistEntry *dep)
+static ErtsDistOutputBuf* clear_de_out_queues(DistEntry* dep)
{
- Sint obufsize = 0;
- ErtsAtomCache *cache;
- ErtsProcList *suspendees;
ErtsDistOutputBuf *obuf;
- erts_smp_de_rwlock(dep);
- cache = dep->cache;
- dep->cache = NULL;
-
-#ifdef DEBUG
- erts_smp_de_links_lock(dep);
- ASSERT(!dep->nlinks);
- ASSERT(!dep->node_links);
- ASSERT(!dep->monitors);
- erts_smp_de_links_unlock(dep);
-#endif
-
- erts_smp_mtx_lock(&dep->qlock);
+ ERTS_LC_ASSERT(erts_lc_mtx_is_locked(&dep->qlock));
if (!dep->out_queue.last)
obuf = dep->finalized_out_queue.first;
@@ -682,21 +705,24 @@ static void clear_dist_entry(DistEntry *dep)
obuf = dep->out_queue.first;
}
+ if (dep->tmp_out_queue.first) {
+ dep->tmp_out_queue.last->next = obuf;
+ obuf = dep->tmp_out_queue.first;
+ }
+
dep->out_queue.first = NULL;
dep->out_queue.last = NULL;
+ dep->tmp_out_queue.first = NULL;
+ dep->tmp_out_queue.last = NULL;
dep->finalized_out_queue.first = NULL;
dep->finalized_out_queue.last = NULL;
- dep->status = 0;
- suspendees = get_suspended_on_de(dep, ERTS_DE_QFLGS_ALL);
-
- erts_smp_mtx_unlock(&dep->qlock);
- erts_smp_atomic_set_nob(&dep->dist_cmd_scheduled, 0);
- dep->send = NULL;
- erts_smp_de_rwunlock(dep);
- erts_resume_processes(suspendees);
+ return obuf;
+}
- delete_cache(cache);
+static void free_de_out_queues(DistEntry* dep, ErtsDistOutputBuf *obuf)
+{
+ Sint obufsize = 0;
while (obuf) {
ErtsDistOutputBuf *fobuf;
@@ -707,13 +733,57 @@ static void clear_dist_entry(DistEntry *dep)
}
if (obufsize) {
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qsize >= obufsize);
- dep->qsize -= obufsize;
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_lock(&dep->qlock);
+ ASSERT(erts_atomic_read_nob(&dep->qsize) >= obufsize);
+ erts_atomic_add_nob(&dep->qsize,
+ (erts_aint_t) -obufsize);
+ erts_mtx_unlock(&dep->qlock);
}
}
+static void clear_dist_entry(DistEntry *dep)
+{
+ ErtsAtomCache *cache;
+ ErtsProcList *suspendees;
+ ErtsDistOutputBuf *obuf;
+
+ erts_de_rwlock(dep);
+ erts_atomic_set_nob(&dep->input_handler,
+ (erts_aint_t) NIL);
+ cache = dep->cache;
+ dep->cache = NULL;
+
+#ifdef DEBUG
+ erts_de_links_lock(dep);
+ ASSERT(!dep->nlinks);
+ ASSERT(!dep->node_links);
+ ASSERT(!dep->monitors);
+ erts_de_links_unlock(dep);
+#endif
+
+ erts_mtx_lock(&dep->qlock);
+
+ erts_atomic64_set_nob(&dep->in, 0);
+ erts_atomic64_set_nob(&dep->out, 0);
+
+ obuf = clear_de_out_queues(dep);
+ dep->status = 0;
+ suspendees = get_suspended_on_de(dep, ERTS_DE_QFLGS_ALL);
+
+ erts_mtx_unlock(&dep->qlock);
+ erts_atomic_set_nob(&dep->dist_cmd_scheduled, 0);
+ dep->send = NULL;
+ erts_de_rwunlock(dep);
+
+ erts_resume_processes(suspendees);
+
+ delete_cache(cache);
+
+ free_de_out_queues(dep, obuf);
+ if (dep->transcode_ctx)
+ transcode_free_ctx(dep);
+}
+
int erts_dsend_context_dtor(Binary* ctx_bin)
{
ErtsSendContext* ctx = ERTS_MAGIC_BIN_DATA(ctx_bin);
@@ -729,8 +799,8 @@ int erts_dsend_context_dtor(Binary* ctx_bin)
if (ctx->dss.phase >= ERTS_DSIG_SEND_PHASE_ALLOC && ctx->dss.obuf) {
free_dist_obuf(ctx->dss.obuf);
}
- if (ctx->dep_to_deref)
- erts_deref_dist_entry(ctx->dep_to_deref);
+ if (ctx->deref_dep)
+ erts_deref_dist_entry(ctx->dep);
return 1;
}
@@ -815,9 +885,9 @@ erts_dsig_send_m_exit(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
watched, watcher, ref, reason);
#ifdef DEBUG
- erts_smp_de_links_lock(dsdp->dep);
+ erts_de_links_lock(dsdp->dep);
ASSERT(!erts_lookup_monitor(dsdp->dep->monitors, ref));
- erts_smp_de_links_unlock(dsdp->dep);
+ erts_de_links_unlock(dsdp->dep);
#endif
res = dsig_send_ctl(dsdp, ctl, 1);
@@ -908,11 +978,30 @@ erts_dsig_send_msg(Eterm remote, Eterm message, ErtsSendContext* ctx)
}
#endif
- if (token != NIL)
- ctl = TUPLE4(&ctx->ctl_heap[0],
- make_small(DOP_SEND_TT), am_Empty, remote, token);
- else
- ctl = TUPLE3(&ctx->ctl_heap[0], make_small(DOP_SEND), am_Empty, remote);
+ if (token != NIL) {
+ Eterm el1, el2;
+ if (ctx->dep->flags & DFLAG_SEND_SENDER) {
+ el1 = make_small(DOP_SEND_SENDER_TT);
+ el2 = sender->common.id;
+ }
+ else {
+ el1 = make_small(DOP_SEND_TT);
+ el2 = am_Empty;
+ }
+ ctl = TUPLE4(&ctx->ctl_heap[0], el1, el2, remote, token);
+ }
+ else {
+ Eterm el1, el2;
+ if (ctx->dep->flags & DFLAG_SEND_SENDER) {
+ el1 = make_small(DOP_SEND_SENDER);
+ el2 = sender->common.id;
+ }
+ else {
+ el1 = make_small(DOP_SEND);
+ el2 = am_Empty;
+ }
+ ctl = TUPLE3(&ctx->ctl_heap[0], el1, el2, remote);
+ }
DTRACE6(message_send, sender_name, receiver_name,
msize, tok_label, tok_lastcnt, tok_serial);
DTRACE7(message_send_remote, sender_name, node_name, receiver_name,
@@ -1149,22 +1238,25 @@ int erts_net_message(Port *prt,
ErtsLink *lnk;
Uint tuple_arity;
int res;
+ Uint32 connection_id;
#ifdef ERTS_DIST_MSG_DBG
ErlDrvSizeT orig_len = len;
#endif
UseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
- ERTS_SMP_CHK_NO_PROC_LOCKS;
+ ERTS_CHK_NO_PROC_LOCKS;
- ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+ ERTS_LC_ASSERT(!prt || erts_lc_is_port_locked(prt));
if (!erts_is_alive) {
UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
return 0;
}
- if (hlen != 0)
- goto data_error;
+
+
+ ASSERT(hlen == 0);
+
if (len == 0) { /* HANDLE TICK !!! */
UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
return 0;
@@ -1183,30 +1275,31 @@ int erts_net_message(Port *prt,
len--;
}
- if (len == 0) {
- PURIFY_MSG("data error");
- goto data_error;
- }
-
- res = erts_prepare_dist_ext(&ede, t, len, dep, dep->cache);
+ res = erts_prepare_dist_ext(&ede, t, len, dep, dep->cache, &connection_id);
- if (res >= 0)
- res = ctl_len = erts_decode_dist_ext_size(&ede);
- else {
+ switch (res) {
+ case ERTS_PREP_DIST_EXT_CLOSED:
+ return 0; /* Connection not alive; ignore signal... */
+ case ERTS_PREP_DIST_EXT_FAILED:
#ifdef ERTS_DIST_MSG_DBG
erts_fprintf(stderr, "DIST MSG DEBUG: erts_prepare_dist_ext() failed:\n");
bw(buf, orig_len);
#endif
- ctl_len = 0;
- }
-
- if (res < 0) {
+ goto data_error;
+ case ERTS_PREP_DIST_EXT_SUCCESS:
+ ctl_len = erts_decode_dist_ext_size(&ede);
+ if (ctl_len < 0) {
#ifdef ERTS_DIST_MSG_DBG
- erts_fprintf(stderr, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
- bw(buf, orig_len);
+ erts_fprintf(stderr, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
+ bw(buf, orig_len);
#endif
- PURIFY_MSG("data error");
- goto data_error;
+ PURIFY_MSG("data error");
+ goto data_error;
+ }
+ break;
+ default:
+ ERTS_INTERNAL_ERROR("Unexpected result from erts_prepare_dist_ext()");
+ break;
}
if (ctl_len > DIST_CTL_DEFAULT_SIZE) {
@@ -1237,6 +1330,7 @@ int erts_net_message(Port *prt,
}
token_size = 0;
+ token = NIL;
switch (type = unsigned_val(tuple[1])) {
case DOP_LINK:
@@ -1257,7 +1351,7 @@ int erts_net_message(Port *prt,
/* This is tricky (we MUST force a distributed send) */
ErtsDSigData dsd;
int code;
- code = erts_dsig_prepare(&dsd, dep, NULL, ERTS_DSP_NO_LOCK, 0);
+ code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
if (code == ERTS_DSIG_PREP_CONNECTED) {
code = erts_dsig_send_exit(&dsd, to, from, am_noproc);
ASSERT(code == ERTS_DSIG_SEND_OK);
@@ -1265,23 +1359,23 @@ int erts_net_message(Port *prt,
break;
}
- erts_smp_de_links_lock(dep);
+ erts_de_links_lock(dep);
res = erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, from);
if (res < 0) {
/* It was already there! Lets skip the rest... */
- erts_smp_de_links_unlock(dep);
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+ erts_de_links_unlock(dep);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
break;
}
lnk = erts_add_or_lookup_link(&(dep->nlinks), LINK_PID, rp->common.id);
erts_add_link(&(ERTS_LINK_ROOT(lnk)), LINK_PID, from);
- erts_smp_de_links_unlock(dep);
+ erts_de_links_unlock(dep);
if (IS_TRACED_FL(rp, F_TRACE_PROCS))
trace_proc(NULL, 0, rp, am_getting_linked, from);
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
break;
case DOP_UNLINK: {
@@ -1307,7 +1401,7 @@ int erts_net_message(Port *prt,
trace_proc(NULL, 0, rp, am_getting_unlinked, from);
}
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
erts_remove_dist_link(&dld, to, from, dep);
erts_destroy_dist_link(&dld);
@@ -1349,7 +1443,7 @@ int erts_net_message(Port *prt,
if (!rp) {
ErtsDSigData dsd;
int code;
- code = erts_dsig_prepare(&dsd, dep, NULL, ERTS_DSP_NO_LOCK, 0);
+ code = erts_dsig_prepare(&dsd, dep, NULL, 0, ERTS_DSP_NO_LOCK, 0, 0);
if (code == ERTS_DSIG_PREP_CONNECTED) {
code = erts_dsig_send_m_exit(&dsd, watcher, watched, ref,
am_noproc);
@@ -1359,11 +1453,11 @@ int erts_net_message(Port *prt,
else {
if (is_atom(watched))
watched = rp->common.id;
- erts_smp_de_links_lock(dep);
+ erts_de_links_lock(dep);
erts_add_monitor(&(dep->monitors), MON_ORIGIN, ref, watched, name);
erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, ref, watcher, name);
- erts_smp_de_links_unlock(dep);
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+ erts_de_links_unlock(dep);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
}
break;
@@ -1385,9 +1479,9 @@ int erts_net_message(Port *prt,
goto invalid_message;
}
- erts_smp_de_links_lock(dep);
+ erts_de_links_lock(dep);
mon = erts_remove_monitor(&(dep->monitors),ref);
- erts_smp_de_links_unlock(dep);
+ erts_de_links_unlock(dep);
/* ASSERT(mon != NULL); can happen in case of broken dist message */
if (mon == NULL) {
break;
@@ -1401,7 +1495,7 @@ int erts_net_message(Port *prt,
break;
}
mon = erts_remove_monitor(&ERTS_P_MONITORS(rp), ref);
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
ASSERT(mon != NULL);
if (mon == NULL) {
break;
@@ -1462,42 +1556,56 @@ int erts_net_message(Port *prt,
erts_queue_dist_message(rp, locks, ede_copy, token, from);
if (locks)
- erts_smp_proc_unlock(rp, locks);
+ erts_proc_unlock(rp, locks);
}
break;
+ case DOP_SEND_SENDER_TT: {
+ Uint xsize;
case DOP_SEND_TT:
+
if (tuple_arity != 4) {
goto invalid_message;
}
-
- token_size = size_object(tuple[4]);
- /* Fall through ... */
+
+ token = tuple[4];
+ token_size = size_object(token);
+ xsize = ERTS_HEAP_FRAG_SIZE(token_size);
+ goto send_common;
+
+ case DOP_SEND_SENDER:
case DOP_SEND:
+
+ token = NIL;
+ xsize = 0;
+ if (tuple_arity != 3)
+ goto invalid_message;
+
+ send_common:
+
/*
- * There is intentionally no testing of the cookie (it is always '')
- * from R9B and onwards.
+ * If DOP_SEND_SENDER or DOP_SEND_SENDER_TT element 2 contains
+ * the sender pid (i.e. DFLAG_SEND_SENDER is set); otherwise,
+ * the atom '' (empty cookie).
*/
+ ASSERT((type == DOP_SEND_SENDER || type == DOP_SEND_SENDER_TT)
+ ? (is_pid(tuple[2]) && (dep->flags & DFLAG_SEND_SENDER))
+ : tuple[2] == am_Empty);
+
#ifdef ERTS_DIST_MSG_DBG
dist_msg_dbg(&ede, "MSG", buf, orig_len);
#endif
- if (type != DOP_SEND_TT && tuple_arity != 3) {
- goto invalid_message;
- }
to = tuple[3];
if (is_not_pid(to)) {
goto invalid_message;
}
rp = erts_proc_lookup(to);
if (rp) {
- Uint xsize = type == DOP_SEND ? 0 : ERTS_HEAP_FRAG_SIZE(token_size);
ErtsProcLocks locks = 0;
ErtsDistExternal *ede_copy;
ede_copy = erts_make_dist_ext_copy(&ede, xsize);
- if (type == DOP_SEND) {
- token = NIL;
- } else {
+ if (is_not_nil(token)) {
ErlHeapFragment *heap_frag;
ErlOffHeap *ohp;
ASSERT(xsize);
@@ -1505,15 +1613,15 @@ int erts_net_message(Port *prt,
ERTS_INIT_HEAP_FRAG(heap_frag, token_size, token_size);
hp = heap_frag->mem;
ohp = &heap_frag->off_heap;
- token = tuple[4];
token = copy_struct(token, token_size, &hp, ohp);
}
- erts_queue_dist_message(rp, locks, ede_copy, token, tuple[2]);
+ erts_queue_dist_message(rp, locks, ede_copy, token, am_Empty);
if (locks)
- erts_smp_proc_unlock(rp, locks);
+ erts_proc_unlock(rp, locks);
}
break;
+ }
case DOP_MONITOR_P_EXIT: {
/* We are monitoring a process on the remote node which dies, we get
@@ -1537,7 +1645,7 @@ int erts_net_message(Port *prt,
goto invalid_message;
}
- erts_smp_de_links_lock(dep);
+ erts_de_links_lock(dep);
sysname = dep->sysname;
mon = erts_remove_monitor(&(dep->monitors), ref);
/*
@@ -1546,7 +1654,7 @@ int erts_net_message(Port *prt,
* removed info about monitor. In this case, do nothing
* and everything will be as it should.
*/
- erts_smp_de_links_unlock(dep);
+ erts_de_links_unlock(dep);
if (mon == NULL) {
break;
}
@@ -1560,7 +1668,7 @@ int erts_net_message(Port *prt,
mon = erts_remove_monitor(&ERTS_P_MONITORS(rp), ref);
if (mon == NULL) {
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
break;
}
UseTmpHeapNoproc(3);
@@ -1571,7 +1679,7 @@ int erts_net_message(Port *prt,
erts_queue_monitor_message(rp, &rp_locks,
ref, am_process, watched, reason);
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
erts_destroy_monitor(mon);
UnUseTmpHeapNoproc(3);
break;
@@ -1633,13 +1741,13 @@ int erts_net_message(Port *prt,
if (xres >= 0 && IS_TRACED_FL(rp, F_TRACE_PROCS)) {
/* We didn't exit the process and it is traced */
if (rp_locks & ERTS_PROC_LOCKS_XSIG_SEND) {
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCKS_XSIG_SEND);
+ erts_proc_unlock(rp, ERTS_PROC_LOCKS_XSIG_SEND);
rp_locks &= ~ERTS_PROC_LOCKS_XSIG_SEND;
}
trace_proc(NULL, 0, rp, am_getting_unlinked, from);
}
}
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
erts_remove_dist_link(&dld, to, from, dep);
if (lnk)
@@ -1681,7 +1789,7 @@ int erts_net_message(Port *prt,
token,
NULL,
0);
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
break;
}
@@ -1699,7 +1807,7 @@ int erts_net_message(Port *prt,
if (!rp)
break;
rp->group_leader = STORE_NC_IN_PROC(rp, from);
- erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_MAIN);
+ erts_proc_unlock(rp, ERTS_PROC_LOCK_MAIN);
break;
default:
@@ -1711,7 +1819,7 @@ int erts_net_message(Port *prt,
erts_free(ERTS_ALC_T_DCTRL_BUF, (void *) ctl);
}
UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
- ERTS_SMP_CHK_NO_PROC_LOCKS;
+ ERTS_CHK_NO_PROC_LOCKS;
return 0;
invalid_message:
{
@@ -1727,8 +1835,8 @@ decode_error:
}
data_error:
UnUseTmpHeapNoproc(DIST_CTL_DEFAULT_SIZE);
- erts_deliver_port_exit(prt, dep->cid, am_killed, 0, 1);
- ERTS_SMP_CHK_NO_PROC_LOCKS;
+ erts_kill_dist_connection(dep, connection_id);
+ ERTS_CHK_NO_PROC_LOCKS;
return -1;
}
@@ -1748,6 +1856,31 @@ static int dsig_send_ctl(ErtsDSigData* dsdp, Eterm ctl, int force_busy)
return ret;
}
+static ERTS_INLINE void
+notify_dist_data(Process *c_p, Eterm pid)
+{
+ Process *rp;
+ ErtsProcLocks rp_locks;
+
+ ASSERT(erts_get_scheduler_data()
+ && !ERTS_SCHEDULER_IS_DIRTY(erts_get_scheduler_data()));
+ ASSERT(is_internal_pid(pid));
+
+ if (c_p && c_p->common.id == pid) {
+ rp = c_p;
+ rp_locks = ERTS_PROC_LOCK_MAIN;
+ }
+ else {
+ rp = erts_proc_lookup(pid);
+ rp_locks = 0;
+ }
+
+ if (rp) {
+ ErtsMessage *mp = erts_alloc_message(0, NULL);
+ erts_queue_message(rp, rp_locks, mp, am_dist_data, am_system);
+ }
+}
+
int
erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
{
@@ -1764,7 +1897,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
if (!ctx->c_p || dsdp->no_suspend)
ctx->force_busy = 1;
- ERTS_SMP_LC_ASSERT(!ctx->c_p
+ ERTS_LC_ASSERT(!ctx->c_p
|| (ERTS_PROC_LOCK_MAIN
== erts_proc_lc_my_proc_locks(ctx->c_p)));
@@ -1773,33 +1906,32 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
if (ctx->flags & DFLAG_DIST_HDR_ATOM_CACHE) {
ctx->acmp = erts_get_atom_cache_map(ctx->c_p);
- ctx->pass_through_size = 0;
+ ctx->max_finalize_prepend = 0;
}
else {
ctx->acmp = NULL;
- ctx->pass_through_size = 1;
+ ctx->max_finalize_prepend = 3;
}
#ifdef ERTS_DIST_MSG_DBG
- erts_fprintf(stderr, ">>%s CTL: %T\n", ctx->pass_through_size ? "P" : " ", ctx->ctl);
- if (is_value(ctx->msg))
- erts_fprintf(stderr, " MSG: %T\n", ctx->msg);
+ erts_fprintf(stderr, ">> CTL: %T\n", ctx->ctl);
+ if (is_value(ctx->msg))
+ erts_fprintf(stderr, " MSG: %T\n", ctx->msg);
#endif
- ctx->data_size = ctx->pass_through_size;
+ ctx->data_size = ctx->max_finalize_prepend;
erts_reset_atom_cache_map(ctx->acmp);
erts_encode_dist_ext_size(ctx->ctl, ctx->flags, ctx->acmp, &ctx->data_size);
- if (is_value(ctx->msg)) {
- ctx->u.sc.wstack.wstart = NULL;
- ctx->u.sc.flags = ctx->flags;
- ctx->u.sc.level = 0;
- ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_SIZE;
- } else {
- ctx->phase = ERTS_DSIG_SEND_PHASE_ALLOC;
- }
- break;
+ if (is_non_value(ctx->msg)) {
+ ctx->phase = ERTS_DSIG_SEND_PHASE_ALLOC;
+ break;
+ }
+ ctx->u.sc.wstack.wstart = NULL;
+ ctx->u.sc.flags = ctx->flags;
+ ctx->u.sc.level = 0;
+ ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_SIZE;
case ERTS_DSIG_SEND_PHASE_MSG_SIZE:
if (erts_encode_dist_ext_size_int(ctx->msg, ctx, &ctx->data_size)) {
retval = ERTS_DSIG_SEND_CONTINUE;
@@ -1814,23 +1946,24 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
ctx->data_size += ctx->dhdr_ext_size;
ctx->obuf = alloc_dist_obuf(ctx->data_size);
- ctx->obuf->ext_endp = &ctx->obuf->data[0] + ctx->pass_through_size + ctx->dhdr_ext_size;
+ ctx->obuf->ext_endp = &ctx->obuf->data[0] + ctx->max_finalize_prepend + ctx->dhdr_ext_size;
/* Encode internal version of dist header */
ctx->obuf->extp = erts_encode_ext_dist_header_setup(ctx->obuf->ext_endp, ctx->acmp);
/* Encode control message */
erts_encode_dist_ext(ctx->ctl, &ctx->obuf->ext_endp, ctx->flags, ctx->acmp, NULL, NULL);
- if (is_value(ctx->msg)) {
- ctx->u.ec.flags = ctx->flags;
- ctx->u.ec.level = 0;
- ctx->u.ec.wstack.wstart = NULL;
- ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_ENCODE;
- } else {
- ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
- }
- break;
+ if (is_non_value(ctx->msg)) {
+ ctx->obuf->msg_start = NULL;
+ ctx->phase = ERTS_DSIG_SEND_PHASE_FIN;
+ break;
+ }
+ ctx->u.ec.flags = ctx->flags;
+ ctx->u.ec.level = 0;
+ ctx->u.ec.wstack.wstart = NULL;
+ ctx->obuf->msg_start = ctx->obuf->ext_endp;
- case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
+ ctx->phase = ERTS_DSIG_SEND_PHASE_MSG_ENCODE;
+ case ERTS_DSIG_SEND_PHASE_MSG_ENCODE:
if (erts_encode_dist_ext(ctx->msg, &ctx->obuf->ext_endp, ctx->flags, ctx->acmp, &ctx->u.ec, &ctx->reds)) {
retval = ERTS_DSIG_SEND_CONTINUE;
goto done;
@@ -1843,7 +1976,7 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
int resume = 0;
ASSERT(ctx->obuf->extp < ctx->obuf->ext_endp);
- ASSERT(&ctx->obuf->data[0] <= ctx->obuf->extp - ctx->pass_through_size);
+ ASSERT(&ctx->obuf->data[0] <= ctx->obuf->extp - ctx->max_finalize_prepend);
ASSERT(ctx->obuf->ext_endp <= &ctx->obuf->data[0] + ctx->data_size);
ctx->data_size = ctx->obuf->ext_endp - ctx->obuf->extp;
@@ -1853,28 +1986,48 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
* and if so enqueue the signal and schedule it for send.
*/
ctx->obuf->next = NULL;
- erts_smp_de_rlock(dep);
+ erts_de_rlock(dep);
cid = dep->cid;
- if (cid != dsdp->cid
- || dep->connection_id != dsdp->connection_id
- || dep->status & ERTS_DE_SFLG_EXITING) {
+ if (!(dep->status & (ERTS_DE_SFLG_PENDING | ERTS_DE_SFLG_CONNECTED))
+ || dep->status & ERTS_DE_SFLG_EXITING
+ || dep->connection_id != dsdp->connection_id) {
/* Not the same connection as when we started; drop message... */
- erts_smp_de_runlock(dep);
+ erts_de_runlock(dep);
free_dist_obuf(ctx->obuf);
}
else {
+ Sint qsize;
+ erts_aint32_t qflgs;
ErtsProcList *plp = NULL;
- erts_smp_mtx_lock(&dep->qlock);
- dep->qsize += size_obuf(ctx->obuf);
- if (dep->qsize >= erts_dist_buf_busy_limit)
- dep->qflgs |= ERTS_DE_QFLG_BUSY;
- if (!ctx->force_busy && (dep->qflgs & ERTS_DE_QFLG_BUSY)) {
- erts_smp_mtx_unlock(&dep->qlock);
+ Eterm notify_proc = NIL;
+ Sint obsz = size_obuf(ctx->obuf);
+
+ erts_mtx_lock(&dep->qlock);
+ qsize = erts_atomic_add_read_nob(&dep->qsize, (erts_aint_t) obsz);
+ ASSERT(qsize >= obsz);
+ qflgs = erts_atomic32_read_nob(&dep->qflgs);
+ if (!(qflgs & ERTS_DE_QFLG_BUSY) && qsize >= erts_dist_buf_busy_limit) {
+ erts_atomic32_read_bor_relb(&dep->qflgs, ERTS_DE_QFLG_BUSY);
+ qflgs |= ERTS_DE_QFLG_BUSY;
+ }
+ if (qsize == obsz && (qflgs & ERTS_DE_QFLG_REQ_INFO)) {
+ /* Previously empty queue and info requested... */
+ qflgs = erts_atomic32_read_band_mb(&dep->qflgs,
+ ~ERTS_DE_QFLG_REQ_INFO);
+ if (qflgs & ERTS_DE_QFLG_REQ_INFO) {
+ notify_proc = dep->cid;
+ ASSERT(is_internal_pid(notify_proc));
+ }
+ /* else: requester will send itself the message... */
+ qflgs &= ~ERTS_DE_QFLG_REQ_INFO;
+ }
+ if (!ctx->force_busy && (qflgs & ERTS_DE_QFLG_BUSY)) {
+ erts_mtx_unlock(&dep->qlock);
plp = erts_proclist_create(ctx->c_p);
erts_suspend(ctx->c_p, ERTS_PROC_LOCK_MAIN, NULL);
suspended = 1;
- erts_smp_mtx_lock(&dep->qlock);
+ erts_mtx_lock(&dep->qlock);
}
/* Enqueue obuf on dist entry */
@@ -1885,7 +2038,8 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
dep->out_queue.last = ctx->obuf;
if (!ctx->force_busy) {
- if (!(dep->qflgs & ERTS_DE_QFLG_BUSY)) {
+ qflgs = erts_atomic32_read_nob(&dep->qflgs);
+ if (!(qflgs & ERTS_DE_QFLG_BUSY)) {
if (suspended)
resume = 1; /* was busy when we started, but isn't now */
#ifdef USE_VM_PROBES
@@ -1909,9 +2063,17 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
}
}
- erts_smp_mtx_unlock(&dep->qlock);
- erts_schedule_dist_command(NULL, dep);
- erts_smp_de_runlock(dep);
+ erts_mtx_unlock(&dep->qlock);
+ if (!(dep->status & ERTS_DE_SFLG_PENDING)) {
+ if (is_internal_port(dep->cid))
+ erts_schedule_dist_command(NULL, dep);
+ }
+ else {
+ notify_proc = NIL;
+ }
+ erts_de_runlock(dep);
+ if (is_internal_pid(notify_proc))
+ notify_dist_data(ctx->c_p, notify_proc);
if (resume) {
erts_resume(ctx->c_p, ERTS_PROC_LOCK_MAIN);
@@ -1965,16 +2127,20 @@ static Uint
dist_port_command(Port *prt, ErtsDistOutputBuf *obuf)
{
int fpe_was_unmasked;
- Uint size = obuf->ext_endp - obuf->extp;
+ ErlDrvSizeT size;
+ char *bufp;
- ERTS_SMP_CHK_NO_PROC_LOCKS;
- ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+ ERTS_CHK_NO_PROC_LOCKS;
+ ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
- if (size > (Uint) INT_MAX)
- erts_exit(ERTS_DUMP_EXIT,
- "Absurdly large distribution output data buffer "
- "(%beu bytes) passed.\n",
- size);
+ if (!obuf) {
+ size = 0;
+ bufp = NULL;
+ }
+ else {
+ size = obuf->ext_endp - obuf->extp;
+ bufp = (char*) obuf->extp;
+ }
#ifdef USE_VM_PROBES
if (DTRACE_ENABLED(dist_output)) {
@@ -1989,11 +2155,10 @@ dist_port_command(Port *prt, ErtsDistOutputBuf *obuf)
remote_str, size);
}
#endif
+
prt->caller = NIL;
fpe_was_unmasked = erts_block_fpe();
- (*prt->drv_ptr->output)((ErlDrvData) prt->drv_data,
- (char*) obuf->extp,
- (int) size);
+ (*prt->drv_ptr->output)((ErlDrvData) prt->drv_data, bufp, size);
erts_unblock_fpe(fpe_was_unmasked);
return size;
}
@@ -2002,33 +2167,41 @@ static Uint
dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
{
int fpe_was_unmasked;
- Uint size = obuf->ext_endp - obuf->extp;
+ ErlDrvSizeT size;
SysIOVec iov[2];
ErlDrvBinary* bv[2];
ErlIOVec eiov;
- ERTS_SMP_CHK_NO_PROC_LOCKS;
- ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
-
- if (size > (Uint) INT_MAX)
- erts_exit(ERTS_DUMP_EXIT,
- "Absurdly large distribution output data buffer "
- "(%beu bytes) passed.\n",
- size);
+ ERTS_CHK_NO_PROC_LOCKS;
+ ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
iov[0].iov_base = NULL;
iov[0].iov_len = 0;
bv[0] = NULL;
- iov[1].iov_base = obuf->extp;
- iov[1].iov_len = size;
- bv[1] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
+ if (!obuf) {
+ size = 0;
+ eiov.vsize = 1;
+ }
+ else {
+ size = obuf->ext_endp - obuf->extp;
+ eiov.vsize = 2;
+
+ iov[1].iov_base = obuf->extp;
+ iov[1].iov_len = size;
+ bv[1] = Binary2ErlDrvBinary(ErtsDistOutputBuf2Binary(obuf));
+ }
- eiov.vsize = 2;
eiov.size = size;
eiov.iov = iov;
eiov.binv = bv;
+ if (size > (Uint) INT_MAX)
+ erts_exit(ERTS_DUMP_EXIT,
+ "Absurdly large distribution output data buffer "
+ "(%beu bytes) passed.\n",
+ size);
+
ASSERT(prt->drv_ptr->outputv);
#ifdef USE_VM_PROBES
@@ -2062,7 +2235,6 @@ dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
#endif
#define ERTS_PORT_REDS_DIST_CMD_START 5
-#define ERTS_PORT_REDS_DIST_CMD_FINALIZE 3
#define ERTS_PORT_REDS_DIST_CMD_EXIT 200
#define ERTS_PORT_REDS_DIST_CMD_RESUMED 5
#define ERTS_PORT_REDS_DIST_CMD_DATA(SZ) \
@@ -2071,37 +2243,36 @@ dist_port_commandv(Port *prt, ErtsDistOutputBuf *obuf)
: ((((Sint) (SZ)) >> 10) & ((Sint) ERTS_PORT_REDS_MASK__)))
int
-erts_dist_command(Port *prt, int reds_limit)
+erts_dist_command(Port *prt, int initial_reds)
{
- Sint reds = ERTS_PORT_REDS_DIST_CMD_START;
+ Sint reds = initial_reds - ERTS_PORT_REDS_DIST_CMD_START;
Uint32 status;
Uint32 flags;
- Sint obufsize = 0;
+ Sint qsize, obufsize = 0;
ErtsDistOutputQueue oq, foq;
DistEntry *dep = prt->dist_entry;
Uint (*send)(Port *prt, ErtsDistOutputBuf *obuf);
erts_aint32_t sched_flags;
ErtsSchedulerData *esdp = erts_get_scheduler_data();
- ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+ ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
- erts_smp_refc_inc(&dep->refc, 1); /* Otherwise dist_entry might be
- removed if port command fails */
+ erts_atomic_set_mb(&dep->dist_cmd_scheduled, 0);
- erts_smp_atomic_set_mb(&dep->dist_cmd_scheduled, 0);
-
- erts_smp_de_rlock(dep);
+ erts_de_rlock(dep);
flags = dep->flags;
status = dep->status;
send = dep->send;
- erts_smp_de_runlock(dep);
+ erts_de_runlock(dep);
if (status & ERTS_DE_SFLG_EXITING) {
erts_deliver_port_exit(prt, prt->common.id, am_killed, 0, 1);
- erts_deref_dist_entry(dep);
- return reds + ERTS_PORT_REDS_DIST_CMD_EXIT;
+ reds -= ERTS_PORT_REDS_DIST_CMD_EXIT;
+ return initial_reds - reds;
}
+ ASSERT(!(status & ERTS_DE_SFLG_PENDING));
+
ASSERT(send);
/*
@@ -2112,42 +2283,42 @@ erts_dist_command(Port *prt, int reds_limit)
* a mess.
*/
- erts_smp_mtx_lock(&dep->qlock);
+ erts_mtx_lock(&dep->qlock);
oq.first = dep->out_queue.first;
oq.last = dep->out_queue.last;
dep->out_queue.first = NULL;
dep->out_queue.last = NULL;
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_unlock(&dep->qlock);
foq.first = dep->finalized_out_queue.first;
foq.last = dep->finalized_out_queue.last;
dep->finalized_out_queue.first = NULL;
dep->finalized_out_queue.last = NULL;
- sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
+ sched_flags = erts_atomic32_read_nob(&prt->sched.flags);
- if (reds > reds_limit)
+ if (reds < 0)
goto preempted;
if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT) && foq.first) {
int preempt = 0;
do {
- Uint size;
- ErtsDistOutputBuf *fob;
-
- size = (*send)(prt, foq.first);
- esdp->io.out += (Uint64) size;
+ Uint size;
+ ErtsDistOutputBuf *fob;
+ size = (*send)(prt, foq.first);
+ erts_atomic64_inc_nob(&dep->out);
+ esdp->io.out += (Uint64) size;
#ifdef ERTS_RAW_DIST_MSG_DBG
- erts_fprintf(stderr, ">> ");
- bw(foq.first->extp, size);
+ erts_fprintf(stderr, ">> ");
+ bw(foq.first->extp, size);
#endif
- reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
+ reds -= ERTS_PORT_REDS_DIST_CMD_DATA(size);
fob = foq.first;
obufsize += size_obuf(fob);
foq.first = foq.first->next;
free_dist_obuf(fob);
- sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
- preempt = reds > reds_limit || (sched_flags & ERTS_PTS_FLG_EXIT);
+ sched_flags = erts_atomic32_read_nob(&prt->sched.flags);
+ preempt = reds < 0 || (sched_flags & ERTS_PTS_FLG_EXIT);
if (sched_flags & ERTS_PTS_FLG_BUSY_PORT)
break;
} while (foq.first && !preempt);
@@ -2160,79 +2331,71 @@ erts_dist_command(Port *prt, int reds_limit)
if (sched_flags & ERTS_PTS_FLG_BUSY_PORT) {
if (oq.first) {
ErtsDistOutputBuf *ob;
- int preempt;
+ ErtsDistOutputBuf *last_finalized = NULL;
finalize_only:
- preempt = 0;
ob = oq.first;
ASSERT(ob);
do {
- ob->extp = erts_encode_ext_dist_header_finalize(ob->extp,
- dep->cache,
- flags);
- if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
- *--ob->extp = PASS_THROUGH; /* Old node; 'pass through'
- needed */
- ASSERT(&ob->data[0] <= ob->extp && ob->extp < ob->ext_endp);
- reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
- preempt = reds > reds_limit;
- if (preempt)
- break;
- ob = ob->next;
+ reds = erts_encode_ext_dist_header_finalize(ob, dep, flags, reds);
+ if (reds >= 0) {
+ last_finalized = ob;
+ ob = ob->next;
+ }
} while (ob);
- /*
- * At least one buffer was finalized; if we got preempted,
- * ob points to the last buffer that we finalized.
- */
- if (foq.last)
- foq.last->next = oq.first;
- else
- foq.first = oq.first;
- if (!preempt) {
- /* All buffers finalized */
- foq.last = oq.last;
- oq.first = oq.last = NULL;
- }
- else {
- /* Not all buffers finalized; split oq. */
- foq.last = ob;
- oq.first = ob->next;
- if (oq.first)
- ob->next = NULL;
- else
- oq.last = NULL;
- }
- if (preempt)
- goto preempted;
+ if (last_finalized) {
+ /*
+ * At least one buffer was finalized; if we got preempted,
+ * ob points to the next buffer to continue finalize.
+ */
+ if (foq.last)
+ foq.last->next = oq.first;
+ else
+ foq.first = oq.first;
+ foq.last = last_finalized;
+ if (!ob) {
+ /* All buffers finalized */
+ ASSERT(foq.last == oq.last);
+ ASSERT(foq.last->next == NULL);
+ oq.first = oq.last = NULL;
+ }
+ else {
+ /* Not all buffers finalized; split oq. */
+ ASSERT(foq.last->next == ob);
+ foq.last->next = NULL;
+ oq.first = ob;
+ }
+ }
+ if (reds <= 0)
+ goto preempted;
}
}
else {
+ int de_busy;
int preempt = 0;
while (oq.first && !preempt) {
ErtsDistOutputBuf *fob;
Uint size;
- oq.first->extp
- = erts_encode_ext_dist_header_finalize(oq.first->extp,
- dep->cache,
- flags);
- reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
- if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
- *--oq.first->extp = PASS_THROUGH; /* Old node; 'pass through'
- needed */
+ reds = erts_encode_ext_dist_header_finalize(oq.first, dep, flags, reds);
+ if (reds < 0) {
+ preempt = 1;
+ break;
+ }
ASSERT(&oq.first->data[0] <= oq.first->extp
&& oq.first->extp < oq.first->ext_endp);
size = (*send)(prt, oq.first);
+ erts_atomic64_inc_nob(&dep->out);
esdp->io.out += (Uint64) size;
#ifdef ERTS_RAW_DIST_MSG_DBG
- erts_fprintf(stderr, ">> ");
- bw(oq.first->extp, size);
+ erts_fprintf(stderr, ">> ");
+ bw(oq.first->extp, size);
#endif
- reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
+ reds -= ERTS_PORT_REDS_DIST_CMD_DATA(size);
fob = oq.first;
obufsize += size_obuf(fob);
oq.first = oq.first->next;
free_dist_obuf(fob);
- sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
- preempt = reds > reds_limit || (sched_flags & ERTS_PTS_FLG_EXIT);
+ sched_flags = erts_atomic32_read_nob(&prt->sched.flags);
+ preempt = reds <= 0 || (sched_flags & ERTS_PTS_FLG_EXIT);
if ((sched_flags & ERTS_PTS_FLG_BUSY_PORT) && oq.first && !preempt)
goto finalize_only;
}
@@ -2258,23 +2421,24 @@ erts_dist_command(Port *prt, int reds_limit)
* dist entry in a non-busy state and resume suspended
* processes.
*/
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qsize >= obufsize);
- dep->qsize -= obufsize;
+ erts_mtx_lock(&dep->qlock);
+ de_busy = !!(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_BUSY);
+ qsize = (Sint) erts_atomic_add_read_nob(&dep->qsize,
+ (erts_aint_t) -obufsize);
+ ASSERT(qsize >= 0);
obufsize = 0;
if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT)
- && (dep->qflgs & ERTS_DE_QFLG_BUSY)
- && dep->qsize < erts_dist_buf_busy_limit) {
+ && de_busy && qsize < erts_dist_buf_busy_limit) {
ErtsProcList *suspendees;
int resumed;
suspendees = get_suspended_on_de(dep, ERTS_DE_QFLG_BUSY);
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_unlock(&dep->qlock);
resumed = erts_resume_processes(suspendees);
- reds += resumed*ERTS_PORT_REDS_DIST_CMD_RESUMED;
+ reds -= resumed*ERTS_PORT_REDS_DIST_CMD_RESUMED;
}
else
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_unlock(&dep->qlock);
}
ASSERT(!oq.first && !oq.last);
@@ -2283,14 +2447,18 @@ erts_dist_command(Port *prt, int reds_limit)
if (obufsize != 0) {
ASSERT(obufsize > 0);
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qsize >= obufsize);
- dep->qsize -= obufsize;
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_lock(&dep->qlock);
+#ifdef DEBUG
+ qsize = (Sint) erts_atomic_add_read_nob(&dep->qsize,
+ (erts_aint_t) -obufsize);
+ ASSERT(qsize >= 0);
+#else
+ erts_atomic_add_nob(&dep->qsize, (erts_aint_t) -obufsize);
+#endif
+ erts_mtx_unlock(&dep->qlock);
}
- ASSERT(foq.first || !foq.last);
- ASSERT(!foq.first || foq.last);
+ ASSERT(!!foq.first == !!foq.last);
ASSERT(!dep->finalized_out_queue.first);
ASSERT(!dep->finalized_out_queue.last);
@@ -2300,12 +2468,10 @@ erts_dist_command(Port *prt, int reds_limit)
}
/* Avoid wrapping reduction counter... */
- if (reds > INT_MAX/2)
- reds = INT_MAX/2;
-
- erts_deref_dist_entry(dep);
+ if (reds < INT_MIN/2)
+ reds = INT_MIN/2;
- return reds;
+ return initial_reds - reds;
preempted:
/*
@@ -2313,8 +2479,7 @@ erts_dist_command(Port *prt, int reds_limit)
* since last call to driver.
*/
- ASSERT(oq.first || !oq.last);
- ASSERT(!oq.first || oq.last);
+ ASSERT(!!oq.first == !!oq.last);
if (sched_flags & ERTS_PTS_FLG_EXIT) {
/*
@@ -2338,12 +2503,6 @@ erts_dist_command(Port *prt, int reds_limit)
foq.first = NULL;
foq.last = NULL;
-
-#ifdef DEBUG
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qsize == obufsize);
- erts_smp_mtx_unlock(&dep->qlock);
-#endif
}
else {
if (oq.first) {
@@ -2351,14 +2510,14 @@ erts_dist_command(Port *prt, int reds_limit)
* Unhandle buffers need to be put back first
* in out_queue.
*/
- erts_smp_mtx_lock(&dep->qlock);
- dep->qsize -= obufsize;
+ erts_mtx_lock(&dep->qlock);
+ erts_atomic_add_nob(&dep->qsize, -obufsize);
obufsize = 0;
oq.last->next = dep->out_queue.first;
dep->out_queue.first = oq.first;
if (!dep->out_queue.last)
dep->out_queue.last = oq.last;
- erts_smp_mtx_unlock(&dep->qlock);
+ erts_mtx_unlock(&dep->qlock);
}
erts_schedule_dist_command(prt, NULL);
@@ -2366,6 +2525,369 @@ erts_dist_command(Port *prt, int reds_limit)
goto done;
}
+#if 0
+
+int
+dist_data_finalize(Process *c_p, int reds_limit)
+{
+ int reds = 5;
+ DistEntry *dep = ;
+ ErtsDistOutputQueue oq, foq;
+ ErtsDistOutputBuf *ob;
+ int preempt;
+
+
+ erts_mtx_lock(&dep->qlock);
+ flags = dep->flags;
+ oq.first = dep->out_queue.first;
+ oq.last = dep->out_queue.last;
+ dep->out_queue.first = NULL;
+ dep->out_queue.last = NULL;
+ erts_mtx_unlock(&dep->qlock);
+
+ if (!oq.first) {
+ ASSERT(!oq.last);
+ oq.first = dep->tmp_out_queue.first;
+ oq.last = dep->tmp_out_queue.last;
+ }
+ else {
+ ErtsDistOutputBuf *f, *l;
+ ASSERT(oq.last);
+ if (dep->tmp_out_queue.last) {
+ dep->tmp_out_queue.last->next = oq.first;
+ oq.first = dep->tmp_out_queue.first;
+ }
+ }
+
+ if (!oq.first) {
+ /* Nothing to do... */
+ ASSERT(!oq.last);
+ return reds;
+ }
+
+ foq.first = dep->finalized_out_queue.first;
+ foq.last = dep->finalized_out_queue.last;
+
+ preempt = 0;
+ ob = oq.first;
+ ASSERT(ob);
+
+ do {
+ ob->extp = erts_encode_ext_dist_header_finalize(ob->extp,
+ dep->cache,
+ flags);
+ if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE))
+ *--ob->extp = PASS_THROUGH; /* Old node; 'pass through'
+ needed */
+ ASSERT(&ob->data[0] <= ob->extp && ob->extp < ob->ext_endp);
+ reds += ERTS_PORT_REDS_DIST_CMD_FINALIZE;
+ preempt = reds > reds_limit;
+ if (preempt)
+ break;
+ ob = ob->next;
+ } while (ob);
+ /*
+ * At least one buffer was finalized; if we got preempted,
+ * ob points to the last buffer that we finalized.
+ */
+ if (foq.last)
+ foq.last->next = oq.first;
+ else
+ foq.first = oq.first;
+ if (!preempt) {
+ /* All buffers finalized */
+ foq.last = oq.last;
+ oq.first = oq.last = NULL;
+ }
+ else {
+ /* Not all buffers finalized; split oq. */
+ foq.last = ob;
+ oq.first = ob->next;
+ if (oq.first)
+ ob->next = NULL;
+ else
+ oq.last = NULL;
+ }
+
+ dep->finalized_out_queue.first = foq.first;
+ dep->finalized_out_queue.last = foq.last;
+ dep->tmp_out_queue.first = oq.first;
+ dep->tmp_out_queue.last = oq.last;
+
+ return reds;
+}
+
+#endif
+
+BIF_RETTYPE
+dist_ctrl_get_data_notification_1(BIF_ALIST_1)
+{
+ DistEntry *dep = ERTS_PROC_GET_DIST_ENTRY(BIF_P);
+ erts_aint32_t qflgs;
+ erts_aint_t qsize;
+ Eterm receiver = NIL;
+
+ if (!dep)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ if (erts_dhandle_to_dist_entry(BIF_ARG_1) != dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ /*
+ * Caller is the only one that can consume from this queue
+ * and the only one that can set the req-info flag...
+ */
+
+ erts_de_rlock(dep);
+
+ ASSERT(dep->cid == BIF_P->common.id);
+
+ qflgs = erts_atomic32_read_acqb(&dep->qflgs);
+
+ if (!(qflgs & ERTS_DE_QFLG_REQ_INFO)) {
+ qsize = erts_atomic_read_acqb(&dep->qsize);
+ ASSERT(qsize >= 0);
+ if (qsize > 0)
+ receiver = BIF_P->common.id; /* Notify ourselves... */
+ else { /* Empty queue; set req-info flag... */
+ qflgs = erts_atomic32_read_bor_mb(&dep->qflgs,
+ ERTS_DE_QFLG_REQ_INFO);
+ qsize = erts_atomic_read_acqb(&dep->qsize);
+ ASSERT(qsize >= 0);
+ if (qsize > 0) {
+ qflgs = erts_atomic32_read_band_mb(&dep->qflgs,
+ ~ERTS_DE_QFLG_REQ_INFO);
+ if (qflgs & ERTS_DE_QFLG_REQ_INFO)
+ receiver = BIF_P->common.id; /* Notify ourselves... */
+ /* else: someone else will notify us... */
+ }
+ /* else: still empty queue... */
+ }
+ }
+ /* else: Already requested... */
+
+ erts_de_runlock(dep);
+
+ if (is_internal_pid(receiver))
+ notify_dist_data(BIF_P, receiver);
+
+ BIF_RET(am_ok);
+}
+
+BIF_RETTYPE
+dist_ctrl_put_data_2(BIF_ALIST_2)
+{
+ DistEntry *dep;
+ ErlDrvSizeT size;
+ Eterm input_handler;
+
+ if (is_binary(BIF_ARG_2))
+ size = binary_size(BIF_ARG_2);
+ else if (is_nil(BIF_ARG_2))
+ size = 0;
+ else if (is_list(BIF_ARG_2))
+ BIF_TRAP2(dist_ctrl_put_data_trap,
+ BIF_P, BIF_ARG_1, BIF_ARG_2);
+ else
+ BIF_ERROR(BIF_P, BADARG);
+
+ dep = erts_dhandle_to_dist_entry(BIF_ARG_1);
+ if (!dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ input_handler = (Eterm) erts_atomic_read_nob(&dep->input_handler);
+
+ if (input_handler != BIF_P->common.id)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ erts_atomic64_inc_nob(&dep->in);
+
+ if (size != 0) {
+ byte *data, *temp_alloc = NULL;
+
+ data = (byte *) erts_get_aligned_binary_bytes(BIF_ARG_2, &temp_alloc);
+ if (!data)
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
+
+ (void) erts_net_message(NULL, dep, NULL, 0, data, size);
+ /*
+ * We ignore any decode failures. On fatal failures the
+ * connection will be taken down by killing the
+ * distribution channel controller...
+ */
+
+ erts_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
+
+ BUMP_REDS(BIF_P, 5);
+
+ erts_free_aligned_binary_bytes(temp_alloc);
+
+ }
+
+ BIF_RET(am_ok);
+}
+
+BIF_RETTYPE
+dist_get_stat_1(BIF_ALIST_1)
+{
+ Sint64 read, write, pend;
+ Eterm res, *hp, **hpp;
+ Uint sz, *szp;
+ DistEntry *dep = erts_dhandle_to_dist_entry(BIF_ARG_1);
+
+ if (!dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_de_rlock(dep);
+
+ read = (Sint64) erts_atomic64_read_nob(&dep->in);
+ write = (Sint64) erts_atomic64_read_nob(&dep->out);
+ pend = (Sint64) erts_atomic_read_nob(&dep->qsize);
+
+ erts_de_runlock(dep);
+
+ sz = 0;
+ szp = &sz;
+ hpp = NULL;
+
+ while (1) {
+ res = erts_bld_tuple(hpp, szp, 4,
+ am_ok,
+ erts_bld_sint64(hpp, szp, read),
+ erts_bld_sint64(hpp, szp, write),
+ pend ? am_true : am_false);
+ if (hpp)
+ break;
+ hp = HAlloc(BIF_P, sz);
+ hpp = &hp;
+ szp = NULL;
+ }
+
+ BIF_RET(res);
+}
+
+BIF_RETTYPE
+dist_ctrl_input_handler_2(BIF_ALIST_2)
+{
+ DistEntry *dep = ERTS_PROC_GET_DIST_ENTRY(BIF_P);
+
+ if (!dep)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ if (erts_dhandle_to_dist_entry(BIF_ARG_1) != dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ if (is_not_internal_pid(BIF_ARG_2))
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_atomic_set_nob(&dep->input_handler,
+ (erts_aint_t) BIF_ARG_2);
+
+ BIF_RET(am_ok);
+}
+
+BIF_RETTYPE
+dist_ctrl_get_data_1(BIF_ALIST_1)
+{
+ DistEntry *dep = ERTS_PROC_GET_DIST_ENTRY(BIF_P);
+ const Sint initial_reds = ERTS_BIF_REDS_LEFT(BIF_P);
+ Sint reds = initial_reds;
+ ErtsDistOutputBuf *obuf;
+ Eterm *hp;
+ ProcBin *pb;
+ erts_aint_t qsize;
+
+ if (!dep)
+ BIF_ERROR(BIF_P, EXC_NOTSUP);
+
+ if (erts_dhandle_to_dist_entry(BIF_ARG_1) != dep)
+ BIF_ERROR(BIF_P, BADARG);
+
+ erts_de_rlock(dep);
+
+ if (dep->status & ERTS_DE_SFLG_EXITING)
+ goto return_none;
+
+ ASSERT(dep->cid == BIF_P->common.id);
+
+#if 0
+ if (dep->finalized_out_queue.first) {
+ obuf = dep->finalized_out_queue.first;
+ dep->finalized_out_queue.first = obuf->next;
+ if (!obuf->next)
+ dep->finalized_out_queue.last = NULL;
+ }
+ else
+#endif
+ {
+ if (!dep->tmp_out_queue.first) {
+ ASSERT(!dep->tmp_out_queue.last);
+ ASSERT(!dep->transcode_ctx);
+ qsize = erts_atomic_read_acqb(&dep->qsize);
+ if (qsize > 0) {
+ erts_mtx_lock(&dep->qlock);
+ dep->tmp_out_queue.first = dep->out_queue.first;
+ dep->tmp_out_queue.last = dep->out_queue.last;
+ dep->out_queue.first = NULL;
+ dep->out_queue.last = NULL;
+ erts_mtx_unlock(&dep->qlock);
+ }
+ }
+
+ if (!dep->tmp_out_queue.first) {
+ ASSERT(!dep->tmp_out_queue.last);
+ return_none:
+ erts_de_runlock(dep);
+ BIF_RET(am_none);
+ }
+
+ obuf = dep->tmp_out_queue.first;
+ reds = erts_encode_ext_dist_header_finalize(obuf, dep, dep->flags, reds);
+ if (reds < 0) {
+ erts_de_runlock(dep);
+ ERTS_BIF_YIELD1(bif_export[BIF_dist_ctrl_get_data_1],
+ BIF_P, BIF_ARG_1);
+ }
+
+ dep->tmp_out_queue.first = obuf->next;
+ if (!obuf->next)
+ dep->tmp_out_queue.last = NULL;
+ }
+
+ erts_atomic64_inc_nob(&dep->out);
+
+ erts_de_runlock(dep);
+
+ hp = HAlloc(BIF_P, PROC_BIN_SIZE);
+ pb = (ProcBin *) (char *) hp;
+ pb->thing_word = HEADER_PROC_BIN;
+ pb->size = obuf->ext_endp - obuf->extp;
+ pb->next = MSO(BIF_P).first;
+ MSO(BIF_P).first = (struct erl_off_heap_header*) pb;
+ pb->val = ErtsDistOutputBuf2Binary(obuf);
+ pb->bytes = (byte*) obuf->extp;
+ pb->flags = 0;
+
+ qsize = erts_atomic_add_read_nob(&dep->qsize, -size_obuf(obuf));
+ ASSERT(qsize >= 0);
+
+ if (qsize < erts_dist_buf_busy_limit/2
+ && (erts_atomic32_read_acqb(&dep->qflgs) & ERTS_DE_QFLG_BUSY)) {
+ ErtsProcList *resume_procs = NULL;
+ erts_mtx_lock(&dep->qlock);
+ resume_procs = get_suspended_on_de(dep, ERTS_DE_QFLG_BUSY);
+ erts_mtx_unlock(&dep->qlock);
+ if (resume_procs) {
+ int resumed = erts_resume_processes(resume_procs);
+ reds -= resumed*ERTS_PORT_REDS_DIST_CMD_RESUMED;
+ }
+ }
+
+ BIF_RET2(make_binary(pb), (initial_reds - reds));
+}
+
void
erts_dist_port_not_busy(Port *prt)
{
@@ -2385,24 +2907,33 @@ erts_dist_port_not_busy(Port *prt)
erts_schedule_dist_command(prt, NULL);
}
+static void kill_connection(DistEntry *dep)
+{
+ ERTS_LC_ASSERT(erts_lc_is_de_rwlocked(dep));
+ ASSERT(dep->status == ERTS_DE_SFLG_CONNECTED);
+
+ dep->status |= ERTS_DE_SFLG_EXITING;
+ erts_mtx_lock(&dep->qlock);
+ ASSERT(!(erts_atomic32_read_nob(&dep->qflgs) & ERTS_DE_QFLG_EXIT));
+ erts_atomic32_read_bor_nob(&dep->qflgs, ERTS_DE_QFLG_EXIT);
+ erts_mtx_unlock(&dep->qlock);
+
+ if (is_internal_port(dep->cid))
+ erts_schedule_dist_command(NULL, dep);
+ else if (is_internal_pid(dep->cid))
+ schedule_kill_dist_ctrl_proc(dep->cid);
+}
+
void
erts_kill_dist_connection(DistEntry *dep, Uint32 connection_id)
{
- erts_smp_de_rwlock(dep);
- if (is_internal_port(dep->cid)
- && connection_id == dep->connection_id
- && !(dep->status & ERTS_DE_SFLG_EXITING)) {
-
- dep->status |= ERTS_DE_SFLG_EXITING;
+ erts_de_rwlock(dep);
+ if (connection_id == dep->connection_id
+ && dep->status == ERTS_DE_SFLG_CONNECTED) {
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(!(dep->qflgs & ERTS_DE_QFLG_EXIT));
- dep->qflgs |= ERTS_DE_QFLG_EXIT;
- erts_smp_mtx_unlock(&dep->qlock);
-
- erts_schedule_dist_command(NULL, dep);
+ kill_connection(dep);
}
- erts_smp_de_rwunlock(dep);
+ erts_de_rwunlock(dep);
}
struct print_to_data {
@@ -2517,9 +3048,6 @@ info_dist_entry(fmtfn_t to, void *arg, DistEntry *dep, int visible, int connecte
}
erts_print(to, arg, "Name: %T", dep->sysname);
-#ifdef DEBUG
- erts_print(to, arg, " (refc=%d)", erts_smp_refc_read(&dep->refc, 0));
-#endif
erts_print(to, arg, "\n");
if (!connected && is_nil(dep->cid)) {
if (dep->nlinks) {
@@ -2563,6 +3091,10 @@ int distribution_info(fmtfn_t to, void *arg) /* Called by break handler */
info_dist_entry(to, arg, dep, 0, 1);
}
+ for (dep = erts_pending_dist_entries; dep; dep = dep->next) {
+ info_dist_entry(to, arg, dep, 0, 0);
+ }
+
for (dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
if (dep != erts_this_dist_entry) {
info_dist_entry(to, arg, dep, 0, 0);
@@ -2585,7 +3117,6 @@ int distribution_info(fmtfn_t to, void *arg) /* Called by break handler */
monitor_node -- turn on/off node monitoring
node controller only:
- dist_exit/3 -- send exit signals from remote to local process
dist_link/2 -- link a remote process to a local
dist_unlink/2 -- unlink a remote from a local
****************************************************************************/
@@ -2595,15 +3126,6 @@ int distribution_info(fmtfn_t to, void *arg) /* Called by break handler */
/**********************************************************************
** Set the node name of current node fail if node already is set.
** setnode(name@host, Creation)
- ** loads functions pointer to trap_functions from module erlang.
- ** erlang:dsend/2
- ** erlang:dlink/1
- ** erlang:dunlink/1
- ** erlang:dmonitor_node/3
- ** erlang:dgroup_leader/2
- ** erlang:dexit/2
- ** -- are these needed ?
- ** dexit/1
***********************************************************************/
BIF_RETTYPE setnode_2(BIF_ALIST_2)
@@ -2627,44 +3149,51 @@ BIF_RETTYPE setnode_2(BIF_ALIST_2)
goto error;
/* Check that all trap functions are defined !! */
- if (dsend2_trap->addressv[0] == NULL ||
- dsend3_trap->addressv[0] == NULL ||
- /* dsend_nosuspend_trap->address == NULL ||*/
- dlink_trap->addressv[0] == NULL ||
- dunlink_trap->addressv[0] == NULL ||
- dmonitor_node_trap->addressv[0] == NULL ||
- dgroup_leader_trap->addressv[0] == NULL ||
- dmonitor_p_trap->addressv[0] == NULL ||
- dexit_trap->addressv[0] == NULL) {
+ if (dmonitor_node_trap->addressv[0] == NULL ||
+ dmonitor_p_trap->addressv[0] == NULL) {
goto error;
}
- net_kernel = erts_whereis_process(BIF_P, ERTS_PROC_LOCK_MAIN,
- am_net_kernel, ERTS_PROC_LOCK_MAIN, 0);
- if (!net_kernel)
+ net_kernel = erts_whereis_process(BIF_P,
+ ERTS_PROC_LOCK_MAIN,
+ am_net_kernel,
+ ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_STATUS,
+ 0);
+ if (!net_kernel || ERTS_PROC_GET_DIST_ENTRY(net_kernel))
goto error;
/* By setting F_DISTRIBUTION on net_kernel,
- * do_net_exist will be called when net_kernel is terminated !! */
+ * erts_do_net_exits will be called when net_kernel is terminated !! */
net_kernel->flags |= F_DISTRIBUTION;
- if (net_kernel != BIF_P)
- erts_smp_proc_unlock(net_kernel, ERTS_PROC_LOCK_MAIN);
+ erts_proc_unlock(net_kernel,
+ (ERTS_PROC_LOCK_STATUS
+ | ((net_kernel != BIF_P)
+ ? ERTS_PROC_LOCK_MAIN
+ : 0)));
#ifdef DEBUG
- erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
+ erts_rwmtx_rlock(&erts_dist_table_rwmtx);
ASSERT(!erts_visible_dist_entries && !erts_hidden_dist_entries);
- erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
+ erts_rwmtx_runlock(&erts_dist_table_rwmtx);
#endif
- erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
- erts_smp_thr_progress_block();
+ erts_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
+ erts_thr_progress_block();
inc_no_nodes();
erts_set_this_node(BIF_ARG_1, (Uint32) creation);
erts_is_alive = 1;
send_nodes_mon_msgs(NULL, am_nodeup, BIF_ARG_1, am_visible, NIL);
- erts_smp_thr_progress_unblock();
- erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
+ erts_thr_progress_unblock();
+ erts_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
+
+ /*
+ * Note erts_this_dist_entry is changed by erts_set_this_node(),
+ * so we *need* to use the new one after erts_set_this_node()
+ * is called.
+ */
+ erts_ref_dist_entry(erts_this_dist_entry);
+ ERTS_PROC_SET_DIST_ENTRY(net_kernel, erts_this_dist_entry);
BIF_RET(am_true);
@@ -2695,18 +3224,20 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
Eterm ic, oc;
Eterm *tp;
DistEntry *dep = NULL;
+ ErtsProcLocks proc_unlock = 0;
+ Process *proc;
Port *pp = NULL;
-
- /* Prepare for success */
- ERTS_BIF_PREP_RET(ret, am_true);
+ Eterm notify_proc;
+ erts_aint32_t qflgs;
/*
* Check and pick out arguments
*/
if (!is_node_name_atom(BIF_ARG_1) ||
- is_not_internal_port(BIF_ARG_2) ||
- (erts_this_node->sysname == am_Noname)) {
+ !(is_internal_port(BIF_ARG_2)
+ || is_internal_pid(BIF_ARG_2))
+ || (erts_this_node->sysname == am_Noname)) {
goto badarg;
}
@@ -2725,21 +3256,25 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
if (!is_atom(ic) || !is_atom(oc))
goto badarg;
- /* DFLAG_EXTENDED_REFERENCES is compulsory from R9 and forward */
- if (!(DFLAG_EXTENDED_REFERENCES & flags)) {
+ if (~flags & DFLAG_DIST_MANDATORY) {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
erts_dsprintf(dsbufp, "%T", BIF_P->common.id);
if (BIF_P->common.u.alive.reg)
erts_dsprintf(dsbufp, " (%T)", BIF_P->common.u.alive.reg->name);
erts_dsprintf(dsbufp,
" attempted to enable connection to node %T "
- "which is not able to handle extended references.\n",
+ "which does not support all mandatory capabilities.\n",
BIF_ARG_1);
erts_send_error_to_logger(BIF_P->group_leader, dsbufp);
goto badarg;
}
/*
+ * ToDo: Should we not pass connection_id as well
+ * to make sure it's the right connection we commit.
+ */
+
+ /*
* Arguments seem to be in order.
*/
@@ -2750,77 +3285,165 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
else if (!dep)
goto system_limit; /* Should never happen!!! */
- pp = erts_id2port_sflgs(BIF_ARG_2,
- BIF_P,
- ERTS_PROC_LOCK_MAIN,
- ERTS_PORT_SFLGS_INVALID_LOOKUP);
- erts_smp_de_rwlock(dep);
+ if (is_internal_pid(BIF_ARG_2)) {
+ if (BIF_P->common.id == BIF_ARG_2) {
+ proc_unlock = 0;
+ proc = BIF_P;
+ }
+ else {
+ proc_unlock = ERTS_PROC_LOCK_MAIN;
+ proc = erts_pid2proc_not_running(BIF_P, ERTS_PROC_LOCK_MAIN,
+ BIF_ARG_2, proc_unlock);
+ }
+ erts_de_rwlock(dep);
- if (!pp || (erts_atomic32_read_nob(&pp->state)
- & ERTS_PORT_SFLG_EXITING))
- goto badarg;
+ if (!proc)
+ goto badarg;
+ else if (proc == ERTS_PROC_LOCK_BUSY) {
+ proc_unlock = 0;
+ goto yield;
+ }
- if ((pp->drv_ptr->flags & ERL_DRV_FLAG_SOFT_BUSY) == 0)
- goto badarg;
+ erts_proc_lock(proc, ERTS_PROC_LOCK_STATUS);
+ proc_unlock |= ERTS_PROC_LOCK_STATUS;
- if (dep->cid == BIF_ARG_2 && pp->dist_entry == dep)
- goto done; /* Already set */
+ if (ERTS_PROC_GET_DIST_ENTRY(proc)) {
+ if (dep == ERTS_PROC_GET_DIST_ENTRY(proc)
+ && (proc->flags & F_DISTRIBUTION)
+ && dep->cid == BIF_ARG_2) {
+ ERTS_BIF_PREP_RET(ret, erts_make_dhandle(BIF_P, dep));
+ goto done;
+ }
+ goto badarg;
+ }
- if (dep->status & ERTS_DE_SFLG_EXITING) {
- /* Suspend on dist entry waiting for the exit to finish */
- ErtsProcList *plp = erts_proclist_create(BIF_P);
- plp->next = NULL;
- erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, NULL);
- erts_smp_mtx_lock(&dep->qlock);
- erts_proclist_store_last(&dep->suspended, plp);
- erts_smp_mtx_unlock(&dep->qlock);
- goto yield;
- }
+ if (dep->status & ERTS_DE_SFLG_EXITING) {
+ /* Suspend on dist entry waiting for the exit to finish */
+ ErtsProcList *plp = erts_proclist_create(BIF_P);
+ plp->next = NULL;
+ erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, NULL);
+ erts_mtx_lock(&dep->qlock);
+ erts_proclist_store_last(&dep->suspended, plp);
+ erts_mtx_unlock(&dep->qlock);
+ goto yield;
+ }
+ if (dep->status != ERTS_DE_SFLG_PENDING) {
+ if (dep->status == 0)
+ erts_set_dist_entry_pending(dep);
+ else
+ goto badarg;
+ }
- ASSERT(!(dep->status & ERTS_DE_SFLG_EXITING));
+ if (is_not_nil(dep->cid))
+ goto badarg;
- if (pp->dist_entry || is_not_nil(dep->cid))
- goto badarg;
+ proc->flags |= F_DISTRIBUTION;
+ ERTS_PROC_SET_DIST_ENTRY(proc, dep);
- erts_atomic32_read_bor_nob(&pp->state, ERTS_PORT_SFLG_DISTRIBUTION);
+ proc_unlock &= ~ERTS_PROC_LOCK_STATUS;
+ erts_proc_unlock(proc, ERTS_PROC_LOCK_STATUS);
+
+ dep->send = NULL; /* Only for distr ports... */
- /*
- * Dist-ports do not use the "busy port message queue" functionality, but
- * instead use "busy dist entry" functionality.
- */
- {
- ErlDrvSizeT disable = ERL_DRV_BUSY_MSGQ_DISABLED;
- erl_drv_busy_msgq_limits(ERTS_Port2ErlDrvPort(pp), &disable, NULL);
}
+ else {
- pp->dist_entry = dep;
+ pp = erts_id2port_sflgs(BIF_ARG_2,
+ BIF_P,
+ ERTS_PROC_LOCK_MAIN,
+ ERTS_PORT_SFLGS_INVALID_LOOKUP);
+ erts_de_rwlock(dep);
- dep->version = version;
- dep->creation = 0;
+ if (!pp || (erts_atomic32_read_nob(&pp->state)
+ & ERTS_PORT_SFLG_EXITING))
+ goto badarg;
- ASSERT(pp->drv_ptr->outputv || pp->drv_ptr->output);
+ if ((pp->drv_ptr->flags & ERL_DRV_FLAG_SOFT_BUSY) == 0)
+ goto badarg;
-#if 1
- dep->send = (pp->drv_ptr->outputv
- ? dist_port_commandv
- : dist_port_command);
-#else
- dep->send = dist_port_command;
-#endif
- ASSERT(dep->send);
+ if (dep->cid == BIF_ARG_2 && pp->dist_entry == dep) {
+ ERTS_BIF_PREP_RET(ret, erts_make_dhandle(BIF_P, dep));
+ goto done; /* Already set */
+ }
+
+ if (dep->status & ERTS_DE_SFLG_EXITING) {
+ /* Suspend on dist entry waiting for the exit to finish */
+ ErtsProcList *plp = erts_proclist_create(BIF_P);
+ plp->next = NULL;
+ erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, NULL);
+ erts_mtx_lock(&dep->qlock);
+ erts_proclist_store_last(&dep->suspended, plp);
+ erts_mtx_unlock(&dep->qlock);
+ goto yield;
+ }
+ if (dep->status != ERTS_DE_SFLG_PENDING) {
+ if (dep->status == 0)
+ erts_set_dist_entry_pending(dep);
+ else
+ goto badarg;
+ }
+
+ if (pp->dist_entry || is_not_nil(dep->cid))
+ goto badarg;
+
+ erts_atomic32_read_bor_nob(&pp->state, ERTS_PORT_SFLG_DISTRIBUTION);
+
+ pp->dist_entry = dep;
+
+ ASSERT(pp->drv_ptr->outputv || pp->drv_ptr->output);
+
+ dep->send = (pp->drv_ptr->outputv
+ ? dist_port_commandv
+ : dist_port_command);
+ ASSERT(dep->send);
+
+ /*
+ * Dist-ports do not use the "busy port message queue" functionality, but
+ * instead use "busy dist entry" functionality.
+ */
+ {
+ ErlDrvSizeT disable = ERL_DRV_BUSY_MSGQ_DISABLED;
+ erl_drv_busy_msgq_limits(ERTS_Port2ErlDrvPort(pp), &disable, NULL);
+ }
+
+ }
+
+ dep->version = version;
+ dep->creation = 0;
#ifdef DEBUG
- erts_smp_mtx_lock(&dep->qlock);
- ASSERT(dep->qsize == 0);
- erts_smp_mtx_unlock(&dep->qlock);
+ ASSERT(erts_atomic_read_nob(&dep->qsize) == 0
+ || (dep->status & ERTS_DE_SFLG_PENDING));
#endif
- erts_set_dist_entry_connected(dep, BIF_ARG_2, flags);
-
if (flags & DFLAG_DIST_HDR_ATOM_CACHE)
create_cache(dep);
- erts_smp_de_rwunlock(dep);
+ erts_set_dist_entry_connected(dep, BIF_ARG_2, flags);
+
+ notify_proc = NIL;
+ if (erts_atomic_read_nob(&dep->qsize)) {
+ if (is_internal_port(dep->cid)) {
+ erts_schedule_dist_command(NULL, dep);
+ }
+ else {
+ qflgs = erts_atomic32_read_nob(&dep->qflgs);
+ if (qflgs & ERTS_DE_QFLG_REQ_INFO) {
+ qflgs = erts_atomic32_read_band_mb(&dep->qflgs,
+ ~ERTS_DE_QFLG_REQ_INFO);
+ if (qflgs & ERTS_DE_QFLG_REQ_INFO) {
+ notify_proc = dep->cid;
+ ASSERT(is_internal_pid(notify_proc));
+ }
+ }
+ }
+ }
+ erts_de_rwunlock(dep);
+ if (is_internal_pid(notify_proc))
+ notify_dist_data(BIF_P, notify_proc);
+
+ ERTS_BIF_PREP_RET(ret, erts_make_dhandle(BIF_P, dep));
+
dep = NULL; /* inc of refc transferred to port (dist_entry field) */
inc_no_nodes();
@@ -2833,13 +3456,16 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
done:
if (dep && dep != erts_this_dist_entry) {
- erts_smp_de_rwunlock(dep);
+ erts_de_rwunlock(dep);
erts_deref_dist_entry(dep);
}
if (pp)
erts_port_release(pp);
+ if (proc_unlock)
+ erts_proc_unlock(proc, proc_unlock);
+
return ret;
yield:
@@ -2856,83 +3482,168 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
goto done;
}
+BIF_RETTYPE erts_internal_new_connection_1(BIF_ALIST_1)
+{
+ DistEntry* dep;
+ Uint32 conn_id;
+ Eterm* hp;
+ Eterm dhandle;
-/**********************************************************************/
-/* dist_exit(Local, Term, Remote) -> Bool */
+ if (is_not_atom(BIF_ARG_1)) {
+ BIF_ERROR(BIF_P, BADARG);
+ }
+ dep = erts_find_or_insert_dist_entry(BIF_ARG_1);
+
+ if (dep == erts_this_dist_entry) {
+ erts_deref_dist_entry(dep);
+ BIF_ERROR(BIF_P, BADARG);
+ }
+
+ erts_de_rwlock(dep);
-BIF_RETTYPE dist_exit_3(BIF_ALIST_3)
+ if (ERTS_DE_IS_CONNECTED(dep) || dep->status & ERTS_DE_SFLG_PENDING)
+ conn_id = dep->connection_id;
+ else if (dep->status == 0) {
+ erts_set_dist_entry_pending(dep);
+ conn_id = dep->connection_id;
+ }
+ else {
+ ASSERT(dep->status & ERTS_DE_SFLG_EXITING);
+ conn_id = (dep->connection_id + 1) & ERTS_DIST_CON_ID_MASK;
+ }
+ erts_de_rwunlock(dep);
+ hp = HAlloc(BIF_P, 3 + ERTS_MAGIC_REF_THING_SIZE);
+ dhandle = erts_build_dhandle(&hp, &BIF_P->off_heap, dep);
+ erts_deref_dist_entry(dep);
+ BIF_RET(TUPLE2(hp, make_small(conn_id), dhandle));
+}
+
+static Sint abort_connection(DistEntry* dep, Uint32 conn_id)
{
- Eterm local;
- Eterm remote;
- DistEntry *rdep;
+ erts_de_rwlock(dep);
- local = BIF_ARG_1;
- remote = BIF_ARG_3;
+ if (dep->connection_id != conn_id)
+ ;
+ else if (dep->status == ERTS_DE_SFLG_CONNECTED) {
+ kill_connection(dep);
+ }
+ else if (dep->status == ERTS_DE_SFLG_PENDING) {
+ NetExitsContext nec = {dep};
+ ErtsLink *nlinks;
+ ErtsLink *node_links;
+ ErtsMonitor *monitors;
+ ErtsAtomCache *cache;
+ ErtsDistOutputBuf *obuf;
+ ErtsProcList *resume_procs;
+ Sint reds = 0;
+
+ ASSERT(is_nil(dep->cid));
+
+ erts_de_links_lock(dep);
+ monitors = dep->monitors;
+ nlinks = dep->nlinks;
+ node_links = dep->node_links;
+ dep->monitors = NULL;
+ dep->nlinks = NULL;
+ dep->node_links = NULL;
+ erts_de_links_unlock(dep);
+
+ cache = dep->cache;
+ dep->cache = NULL;
+ erts_mtx_lock(&dep->qlock);
+ obuf = dep->out_queue.first;
+ dep->out_queue.first = NULL;
+ dep->out_queue.last = NULL;
+ ASSERT(!dep->tmp_out_queue.first);
+ ASSERT(!dep->finalized_out_queue.first);
+ resume_procs = get_suspended_on_de(dep, ERTS_DE_QFLGS_ALL);
+ erts_mtx_unlock(&dep->qlock);
+ erts_atomic_set_nob(&dep->dist_cmd_scheduled, 0);
+ dep->send = NULL;
+
+ erts_set_dist_entry_not_connected(dep);
+
+ erts_de_rwunlock(dep);
+
+ erts_sweep_monitors(monitors, &doit_monitor_net_exits, &nec);
+ erts_sweep_links(nlinks, &doit_link_net_exits, &nec);
+ erts_sweep_links(node_links, &doit_node_link_net_exits, &nec);
+
+ if (resume_procs) {
+ int resumed = erts_resume_processes(resume_procs);
+ reds += resumed*ERTS_PORT_REDS_DIST_CMD_RESUMED;
+ }
- /* Check that remote is a remote process */
- if (is_not_external_pid(remote))
- goto error;
+ delete_cache(cache);
+ free_de_out_queues(dep, obuf);
+ return reds;
+ }
+ erts_de_rwunlock(dep);
+ return 0;
+}
- rdep = external_dist_entry(remote);
-
- if(rdep == erts_this_dist_entry)
- goto error;
+BIF_RETTYPE erts_internal_abort_connection_2(BIF_ALIST_2)
+{
+ DistEntry* dep;
+ Eterm* tp;
- /* Check that local is local */
- if (is_internal_pid(local)) {
- Process *lp;
- ErtsProcLocks lp_locks;
- if (BIF_P->common.id == local) {
- lp_locks = ERTS_PROC_LOCKS_ALL;
- lp = BIF_P;
- erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCKS_ALL_MINOR);
- }
- else {
- lp_locks = ERTS_PROC_LOCKS_XSIG_SEND;
- lp = erts_pid2proc(BIF_P, ERTS_PROC_LOCK_MAIN,
- local, lp_locks);
- if (!lp) {
- BIF_RET(am_true); /* ignore */
- }
- }
-
- (void) erts_send_exit_signal(BIF_P,
- remote,
- lp,
- &lp_locks,
- BIF_ARG_2,
- NIL,
- NULL,
- 0);
-#ifdef ERTS_SMP
- if (lp == BIF_P)
- lp_locks &= ~ERTS_PROC_LOCK_MAIN;
-#endif
- erts_smp_proc_unlock(lp, lp_locks);
- if (lp == BIF_P) {
- erts_aint32_t state = erts_smp_atomic32_read_acqb(&BIF_P->state);
- /*
- * We may have exited current process and may have to take action.
- */
- if (state & (ERTS_PSFLG_EXITING|ERTS_PSFLG_PENDING_EXIT)) {
-#ifdef ERTS_SMP
- if (state & ERTS_PSFLG_PENDING_EXIT)
- erts_handle_pending_exit(BIF_P, ERTS_PROC_LOCK_MAIN);
-#endif
- ERTS_BIF_EXITED(BIF_P);
- }
- }
+ if (is_not_atom(BIF_ARG_1) || is_not_tuple_arity(BIF_ARG_2, 2)) {
+ BIF_ERROR(BIF_P, BADARG);
}
- else if (is_external_pid(local)
- && external_dist_entry(local) == erts_this_dist_entry) {
- BIF_RET(am_true); /* ignore */
+ tp = tuple_val(BIF_ARG_2);
+ dep = erts_dhandle_to_dist_entry(tp[2]);
+ if (is_not_small(tp[1]) || dep != erts_find_dist_entry(BIF_ARG_1)
+ || dep == erts_this_dist_entry) {
+ BIF_ERROR(BIF_P, BADARG);
+ }
+
+ if (dep) {
+ Sint reds = abort_connection(dep, unsigned_val(tp[1]));
+ BUMP_REDS(BIF_P, reds);
}
- else
- goto error;
BIF_RET(am_true);
+}
- error:
- BIF_ERROR(BIF_P, BADARG);
+int erts_auto_connect(DistEntry* dep, Process *proc, ErtsProcLocks proc_locks)
+{
+ erts_de_rwlock(dep);
+ if (dep->status != 0) {
+ erts_de_rwunlock(dep);
+ }
+ else {
+ Process* net_kernel;
+ ErtsProcLocks nk_locks = ERTS_PROC_LOCK_MSGQ;
+ Eterm *hp;
+ ErlOffHeap *ohp;
+ ErtsMessage *mp;
+ Eterm msg, dhandle;
+ Uint32 conn_id;
+
+ erts_set_dist_entry_pending(dep);
+ conn_id = dep->connection_id;
+ erts_de_rwunlock(dep);
+
+ net_kernel = erts_whereis_process(proc, proc_locks,
+ am_net_kernel, nk_locks, 0);
+ if (!net_kernel) {
+ abort_connection(dep, conn_id);
+ return 0;
+ }
+
+ /*
+ * Send {auto_connect, Node, ConnId, DHandle} to net_kernel
+ */
+ mp = erts_alloc_message_heap(net_kernel, &nk_locks,
+ 5 + ERTS_MAGIC_REF_THING_SIZE,
+ &hp, &ohp);
+ dhandle = erts_build_dhandle(&hp, ohp, dep);
+ msg = TUPLE4(hp, am_auto_connect, dep->sysname, make_small(conn_id),
+ dhandle);
+ erts_queue_message(net_kernel, nk_locks, mp, msg, proc->common.id);
+ erts_proc_unlock(net_kernel, nk_locks);
+ }
+
+ return 1;
}
/**********************************************************************/
@@ -3004,13 +3715,15 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
length = 0;
- erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
+ erts_rwmtx_rlock(&erts_dist_table_rwmtx);
ASSERT(erts_no_of_not_connected_dist_entries > 0);
ASSERT(erts_no_of_hidden_dist_entries >= 0);
+ ASSERT(erts_no_of_pending_dist_entries >= 0);
ASSERT(erts_no_of_visible_dist_entries >= 0);
if(not_connected)
- length += (erts_no_of_not_connected_dist_entries - 1);
+ length += ((erts_no_of_not_connected_dist_entries - 1)
+ + erts_no_of_pending_dist_entries);
if(hidden)
length += erts_no_of_hidden_dist_entries;
if(visible)
@@ -3021,7 +3734,7 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
result = NIL;
if (length == 0) {
- erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
+ erts_rwmtx_runlock(&erts_dist_table_rwmtx);
goto done;
}
@@ -3030,13 +3743,18 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
#ifdef DEBUG
endp = hp + length*2;
#endif
- if(not_connected)
+ if(not_connected) {
for(dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
if (dep != erts_this_dist_entry) {
result = CONS(hp, dep->sysname, result);
hp += 2;
}
+ }
+ for(dep = erts_pending_dist_entries; dep; dep = dep->next) {
+ result = CONS(hp, dep->sysname, result);
+ hp += 2;
}
+ }
if(hidden)
for(dep = erts_hidden_dist_entries; dep; dep = dep->next) {
result = CONS(hp, dep->sysname, result);
@@ -3052,7 +3770,7 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
hp += 2;
}
ASSERT(endp == hp);
- erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
+ erts_rwmtx_runlock(&erts_dist_table_rwmtx);
done:
UnUseTmpHeap(2,BIF_P);
@@ -3081,11 +3799,17 @@ monitor_node(Process* p, Eterm Node, Eterm Bool, Eterm Options)
DistEntry *dep;
ErtsLink *lnk;
Eterm l;
+ int async_connect = 1;
for (l = Options; l != NIL && is_list(l); l = CDR(list_val(l))) {
Eterm t = CAR(list_val(l));
- /* allow_passive_connect the only available option right now */
- if (t != am_allow_passive_connect) {
+ if (t == am_allow_passive_connect) {
+ /*
+ * Handle this horrible feature by falling back on old synchronous
+ * auto-connect (if needed)
+ */
+ async_connect = 0;
+ } else {
BIF_ERROR(p, BADARG);
}
}
@@ -3099,54 +3823,93 @@ monitor_node(Process* p, Eterm Node, Eterm Bool, Eterm Options)
&& (Node != erts_this_node->sysname))) {
BIF_ERROR(p, BADARG);
}
- dep = erts_sysname_to_connected_dist_entry(Node);
- if (!dep) {
- do_trap:
- BIF_TRAP3(dmonitor_node_trap, p, Node, Bool, Options);
- }
- if (dep == erts_this_dist_entry)
- goto done;
-
- erts_smp_proc_lock(p, ERTS_PROC_LOCK_LINK);
- erts_smp_de_rlock(dep);
- if (ERTS_DE_IS_NOT_CONNECTED(dep)) {
- erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
- erts_smp_de_runlock(dep);
- goto do_trap;
- }
- erts_smp_de_links_lock(dep);
- erts_smp_de_runlock(dep);
if (Bool == am_true) {
- ASSERT(dep->cid != NIL);
- lnk = erts_add_or_lookup_link(&(dep->node_links), LINK_NODE,
- p->common.id);
- ++ERTS_LINK_REFC(lnk);
- lnk = erts_add_or_lookup_link(&ERTS_P_LINKS(p), LINK_NODE, Node);
- ++ERTS_LINK_REFC(lnk);
- }
- else {
- lnk = erts_lookup_link(dep->node_links, p->common.id);
- if (lnk != NULL) {
- if ((--ERTS_LINK_REFC(lnk)) == 0) {
- erts_destroy_link(erts_remove_link(&(dep->node_links),
- p->common.id));
- }
- }
- lnk = erts_lookup_link(ERTS_P_LINKS(p), Node);
- if (lnk != NULL) {
- if ((--ERTS_LINK_REFC(lnk)) == 0) {
- erts_destroy_link(erts_remove_link(&ERTS_P_LINKS(p),
- Node));
+ ErtsDSigData dsd;
+ dsd.node = Node;
+ dep = erts_find_or_insert_dist_entry(Node);
+ if (dep == erts_this_dist_entry)
+ goto done;
+
+ erts_proc_lock(p, ERTS_PROC_LOCK_LINK);
+
+ switch (erts_dsig_prepare(&dsd, dep, p,
+ (ERTS_PROC_LOCK_MAIN | ERTS_PROC_LOCK_LINK),
+ ERTS_DSP_RLOCK, 0, async_connect)) {
+ case ERTS_DSIG_PREP_NOT_ALIVE:
+ case ERTS_DSIG_PREP_NOT_CONNECTED:
+ /* Trap to either send 'nodedown' or do passive connection attempt */
+ trap:
+ erts_proc_unlock(p, ERTS_PROC_LOCK_LINK);
+ erts_deref_dist_entry(dep);
+ BIF_TRAP3(dmonitor_node_trap, p, Node, Bool, Options);
+ case ERTS_DSIG_PREP_PENDING:
+ if (!async_connect) {
+ /*
+ * Pending connection may fail, so we must trap
+ * to ensure passive connection attempt
+ */
+ erts_de_runlock(dep);
+ goto trap;
}
- }
+ /*fall through*/
+ case ERTS_DSIG_PREP_CONNECTED:
+ erts_de_links_lock(dep);
+ erts_de_runlock(dep);
+ lnk = erts_add_or_lookup_link(&(dep->node_links), LINK_NODE,
+ p->common.id);
+ ++ERTS_LINK_REFC(lnk);
+ lnk = erts_add_or_lookup_link(&ERTS_P_LINKS(p), LINK_NODE, Node);
+ ++ERTS_LINK_REFC(lnk);
+ erts_de_links_unlock(dep);
+ break;
+ default:
+ ERTS_ASSERT(! "Invalid dsig prepare result");
+ }
+ erts_deref_dist_entry(dep);
+ }
+ else { /* Bool == false */
+ dep = erts_sysname_to_connected_dist_entry(Node);
+ if (!dep) {
+ /*
+ * Before OTP-21 this case triggered auto-connect
+ * and a 'nodedown' message if that failed.
+ * Now it's a simple no-op which feels more reasonable.
+ */
+ BIF_RET(am_true);
+ }
+ if (dep == erts_this_dist_entry)
+ goto done;
+
+ erts_proc_lock(p, ERTS_PROC_LOCK_LINK);
+ erts_de_rlock(dep);
+ if (!(dep->status & (ERTS_DE_SFLG_PENDING | ERTS_DE_SFLG_CONNECTED))) {
+ erts_proc_unlock(p, ERTS_PROC_LOCK_LINK);
+ erts_de_runlock(dep);
+ goto done;
+ }
+ erts_de_links_lock(dep);
+ erts_de_runlock(dep);
+ lnk = erts_lookup_link(dep->node_links, p->common.id);
+ if (lnk != NULL) {
+ if ((--ERTS_LINK_REFC(lnk)) == 0) {
+ erts_destroy_link(erts_remove_link(&(dep->node_links),
+ p->common.id));
+ }
+ }
+ lnk = erts_lookup_link(ERTS_P_LINKS(p), Node);
+ if (lnk != NULL) {
+ if ((--ERTS_LINK_REFC(lnk)) == 0) {
+ erts_destroy_link(erts_remove_link(&ERTS_P_LINKS(p),
+ Node));
+ }
+ }
+ erts_de_links_unlock(dep);
}
- erts_smp_de_links_unlock(dep);
- erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
+ erts_proc_unlock(p, ERTS_PROC_LOCK_LINK);
done:
- erts_deref_dist_entry(dep);
BIF_RET(am_true);
}
@@ -3175,9 +3938,9 @@ BIF_RETTYPE net_kernel_dflag_unicode_io_1(BIF_ALIST_1)
if (de == erts_this_dist_entry) {
BIF_RET(am_true);
}
- erts_smp_de_rlock(de);
+ erts_de_rlock(de);
f = de->flags;
- erts_smp_de_runlock(de);
+ erts_de_runlock(de);
BIF_RET(((f & DFLAG_UNICODE_IO) ? am_true : am_false));
}
@@ -3207,7 +3970,7 @@ struct ErtsNodesMonitor_ {
Uint16 no;
};
-static erts_smp_mtx_t nodes_monitors_mtx;
+static erts_mtx_t nodes_monitors_mtx;
static ErtsNodesMonitor *nodes_monitors;
static ErtsNodesMonitor *nodes_monitors_end;
@@ -3225,7 +3988,7 @@ static ErtsNodesMonitor *nodes_monitors_end;
static void
init_nodes_monitors(void)
{
- erts_smp_mtx_init(&nodes_monitors_mtx, "nodes_monitors", NIL,
+ erts_mtx_init(&nodes_monitors_mtx, "nodes_monitors", NIL,
ERTS_LOCK_FLAGS_PROPERTY_STATIC | ERTS_LOCK_FLAGS_CATEGORY_DISTRIBUTION);
nodes_monitors = NULL;
nodes_monitors_end = NULL;
@@ -3351,10 +4114,10 @@ send_nodes_mon_msgs(Process *c_p, Eterm what, Eterm node, Eterm type, Eterm reas
}
#endif
- ERTS_SMP_LC_ASSERT(!c_p
+ ERTS_LC_ASSERT(!c_p
|| (erts_proc_lc_my_proc_locks(c_p)
== ERTS_PROC_LOCK_MAIN));
- erts_smp_mtx_lock(&nodes_monitors_mtx);
+ erts_mtx_lock(&nodes_monitors_mtx);
for (nmp = nodes_monitors; nmp; nmp = nmp->next) {
int i;
@@ -3387,7 +4150,7 @@ send_nodes_mon_msgs(Process *c_p, Eterm what, Eterm node, Eterm type, Eterm reas
if (rp) {
if (rp == c_p)
rp_locks &= ~ERTS_PROC_LOCK_MAIN;
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
rp = nmp->proc;
@@ -3414,10 +4177,10 @@ send_nodes_mon_msgs(Process *c_p, Eterm what, Eterm node, Eterm type, Eterm reas
if (rp) {
if (rp == c_p)
rp_locks &= ~ERTS_PROC_LOCK_MAIN;
- erts_smp_proc_unlock(rp, rp_locks);
+ erts_proc_unlock(rp, rp_locks);
}
- erts_smp_mtx_unlock(&nodes_monitors_mtx);
+ erts_mtx_unlock(&nodes_monitors_mtx);
}
static Eterm
@@ -3427,8 +4190,8 @@ insert_nodes_monitor(Process *c_p, Uint32 opts)
Eterm res = am_false;
ErtsNodesMonitor *xnmp, *nmp;
- ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&nodes_monitors_mtx));
- ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
+ ERTS_LC_ASSERT(erts_lc_mtx_is_locked(&nodes_monitors_mtx));
+ ERTS_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
xnmp = c_p->nodes_monitors;
if (xnmp) {
@@ -3512,8 +4275,8 @@ remove_nodes_monitors(Process *c_p, Uint32 opts, int all)
Eterm res = am_false;
ErtsNodesMonitor *nmp;
- ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&nodes_monitors_mtx));
- ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
+ ERTS_LC_ASSERT(erts_lc_mtx_is_locked(&nodes_monitors_mtx));
+ ERTS_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) & ERTS_PROC_LOCK_MAIN);
nmp = c_p->nodes_monitors;
ASSERT(!nmp || !nmp->prev || nmp->prev->proc != c_p);
@@ -3555,23 +4318,23 @@ remove_nodes_monitors(Process *c_p, Uint32 opts, int all)
void
erts_delete_nodes_monitors(Process *c_p, ErtsProcLocks locks)
{
-#if defined(ERTS_ENABLE_LOCK_CHECK) && defined(ERTS_SMP)
+#if defined(ERTS_ENABLE_LOCK_CHECK)
if (c_p) {
ErtsProcLocks might_unlock = locks & ~ERTS_PROC_LOCK_MAIN;
if (might_unlock)
erts_proc_lc_might_unlock(c_p, might_unlock);
}
#endif
- if (erts_smp_mtx_trylock(&nodes_monitors_mtx) == EBUSY) {
+ if (erts_mtx_trylock(&nodes_monitors_mtx) == EBUSY) {
ErtsProcLocks unlock_locks = locks & ~ERTS_PROC_LOCK_MAIN;
if (c_p && unlock_locks)
- erts_smp_proc_unlock(c_p, unlock_locks);
- erts_smp_mtx_lock(&nodes_monitors_mtx);
+ erts_proc_unlock(c_p, unlock_locks);
+ erts_mtx_lock(&nodes_monitors_mtx);
if (c_p && unlock_locks)
- erts_smp_proc_lock(c_p, unlock_locks);
+ erts_proc_lock(c_p, unlock_locks);
}
remove_nodes_monitors(c_p, 0, 1);
- erts_smp_mtx_unlock(&nodes_monitors_mtx);
+ erts_mtx_unlock(&nodes_monitors_mtx);
}
Eterm
@@ -3582,7 +4345,7 @@ erts_monitor_nodes(Process *c_p, Eterm on, Eterm olist)
Uint16 opts = (Uint16) 0;
ASSERT(c_p);
- ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
+ ERTS_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
if (on != am_true && on != am_false)
return THE_NON_VALUE;
@@ -3638,14 +4401,14 @@ erts_monitor_nodes(Process *c_p, Eterm on, Eterm olist)
return THE_NON_VALUE;
}
- erts_smp_mtx_lock(&nodes_monitors_mtx);
+ erts_mtx_lock(&nodes_monitors_mtx);
if (on == am_true)
res = insert_nodes_monitor(c_p, opts);
else
res = remove_nodes_monitors(c_p, opts, 0);
- erts_smp_mtx_unlock(&nodes_monitors_mtx);
+ erts_mtx_unlock(&nodes_monitors_mtx);
return res;
}
@@ -3668,8 +4431,8 @@ erts_processes_monitoring_nodes(Process *c_p)
#endif
ASSERT(c_p);
- ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
- erts_smp_mtx_lock(&nodes_monitors_mtx);
+ ERTS_LC_ASSERT(erts_proc_lc_my_proc_locks(c_p) == ERTS_PROC_LOCK_MAIN);
+ erts_mtx_lock(&nodes_monitors_mtx);
sz = 0;
szp = &sz;
@@ -3718,7 +4481,7 @@ erts_processes_monitoring_nodes(Process *c_p)
ASSERT(hp == hend);
- erts_smp_mtx_unlock(&nodes_monitors_mtx);
+ erts_mtx_unlock(&nodes_monitors_mtx);
return res;
}