aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam/erl_process.c
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam/erl_process.c')
-rw-r--r--erts/emulator/beam/erl_process.c198
1 files changed, 185 insertions, 13 deletions
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 138acfeb2c..95d408f79d 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -42,6 +42,7 @@
#include "erl_thr_progress.h"
#include "erl_thr_queue.h"
#include "erl_async.h"
+#include "dtrace-wrapper.h"
#define ERTS_RUNQ_CHECK_BALANCE_REDS_PER_SCHED (2000*CONTEXT_REDS)
#define ERTS_RUNQ_CALL_CHECK_BALANCE_REDS \
@@ -104,6 +105,9 @@ do { \
#define ERTS_EMPTY_RUNQ(RQ) \
((RQ)->len == 0 && (RQ)->misc.start == NULL)
+#define ERTS_EMPTY_RUNQ_PORTS(RQ) \
+ ((RQ)->ports.info.len == 0 && (RQ)->misc.start == NULL)
+
extern BeamInstr beam_apply[];
extern BeamInstr beam_exit[];
extern BeamInstr beam_continue_exit[];
@@ -366,6 +370,9 @@ dbg_chk_aux_work_val(erts_aint32_t value)
#ifdef ERTS_SMP_SCHEDULERS_NEED_TO_CHECK_CHILDREN
valid |= ERTS_SSI_AUX_WORK_CHECK_CHILDREN;
#endif
+#ifdef ERTS_SSI_AUX_WORK_REAP_PORTS
+ valid |= ERTS_SSI_AUX_WORK_REAP_PORTS;
+#endif
if (~valid & value)
erl_exit(ERTS_ABORT_EXIT,
@@ -672,7 +679,11 @@ reply_sched_wall_time(void *vswtrp)
hpp = &hp;
}
- erts_queue_message(rp, &rp_locks, bp, msg, NIL);
+ erts_queue_message(rp, &rp_locks, bp, msg, NIL
+#ifdef USE_VM_PROBES
+ , NIL
+#endif
+ );
if (swtrp->req_sched == esdp->no)
rp_locks &= ~ERTS_PROC_LOCK_MAIN;
@@ -861,8 +872,6 @@ set_aux_work_flags_wakeup_nob(ErtsSchedulerSleepInfo *ssi,
}
}
-#if 0 /* Currently not used */
-
static ERTS_INLINE void
set_aux_work_flags_wakeup_relb(ErtsSchedulerSleepInfo *ssi,
erts_aint32_t flgs)
@@ -882,8 +891,6 @@ set_aux_work_flags_wakeup_relb(ErtsSchedulerSleepInfo *ssi,
}
}
-#endif
-
static ERTS_INLINE erts_aint32_t
set_aux_work_flags(ErtsSchedulerSleepInfo *ssi, erts_aint32_t flgs)
{
@@ -1351,6 +1358,65 @@ handle_check_children(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
#endif
+static void
+notify_reap_ports_relb(void)
+{
+ int i;
+ for (i = 0; i < erts_no_schedulers; i++) {
+ set_aux_work_flags_wakeup_relb(ERTS_SCHED_SLEEP_INFO_IX(i),
+ ERTS_SSI_AUX_WORK_REAP_PORTS);
+ }
+}
+
+erts_smp_atomic32_t erts_halt_progress;
+int erts_halt_code;
+
+static ERTS_INLINE erts_aint32_t
+handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+{
+ unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_REAP_PORTS);
+ awdp->esdp->run_queue->halt_in_progress = 1;
+ if (erts_smp_atomic32_dec_read_acqb(&erts_halt_progress) == 0) {
+ int i;
+ erts_smp_atomic32_set_nob(&erts_halt_progress, 1);
+ for (i = 0; i < erts_max_ports; i++) {
+ Port *prt = &erts_port[i];
+ erts_smp_port_state_lock(prt);
+ if ((prt->status & (ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP
+ | ERTS_PORT_SFLG_HALT))) {
+ erts_smp_port_state_unlock(prt);
+ continue;
+ }
+ /* We need to set the halt flag - get the port lock */
+#ifdef ERTS_SMP
+ erts_smp_atomic_inc_nob(&prt->refc);
+#endif
+ erts_smp_port_state_unlock(prt);
+#ifdef ERTS_SMP
+ erts_smp_mtx_lock(prt->lock);
+#endif
+ if ((prt->status & (ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP
+ | ERTS_PORT_SFLG_HALT))) {
+ erts_port_release(prt);
+ continue;
+ }
+ erts_port_status_bor_set(prt, ERTS_PORT_SFLG_HALT);
+ erts_smp_atomic32_inc_nob(&erts_halt_progress);
+ if (prt->status & (ERTS_PORT_SFLG_EXITING
+ | ERTS_PORT_SFLG_CLOSING)) {
+ erts_port_release(prt);
+ continue;
+ }
+ erts_do_exit_port(prt, prt->id, am_killed);
+ erts_port_release(prt);
+ }
+ if (erts_smp_atomic32_dec_read_nob(&erts_halt_progress) == 0) {
+ erl_exit_flush_async(erts_halt_code, "");
+ }
+ }
+ return aux_work & ~ERTS_SSI_AUX_WORK_REAP_PORTS;
+}
+
#if HAVE_ERTS_MSEG
static ERTS_INLINE erts_aint32_t
@@ -1451,6 +1517,9 @@ handle_aux_work(ErtsAuxWorkData *awdp, erts_aint32_t orig_aux_work)
handle_mseg_cache_check);
#endif
+ HANDLE_AUX_WORK(ERTS_SSI_AUX_WORK_REAP_PORTS,
+ handle_reap_ports);
+
ERTS_DBG_CHK_AUX_WORK_VAL(aux_work);
return aux_work;
@@ -2716,6 +2785,9 @@ try_steal_task_from_victim(ErtsRunQueue *rq, int *rq_lockedp, ErtsRunQueue *vrq)
ERTS_SMP_LC_CHK_RUNQ_LOCK(rq, *rq_lockedp);
ERTS_SMP_LC_CHK_RUNQ_LOCK(vrq, vrq_locked);
+ if (rq->halt_in_progress)
+ goto try_steal_port;
+
/*
* Check for a runnable process to steal...
*/
@@ -2802,6 +2874,8 @@ try_steal_task_from_victim(ErtsRunQueue *rq, int *rq_lockedp, ErtsRunQueue *vrq)
vrq_locked = 1;
}
+ try_steal_port:
+
ERTS_SMP_LC_CHK_RUNQ_LOCK(rq, *rq_lockedp);
ERTS_SMP_LC_CHK_RUNQ_LOCK(vrq, vrq_locked);
@@ -2917,7 +2991,8 @@ try_steal_task(ErtsRunQueue *rq)
erts_smp_runq_lock(rq);
if (!res)
- res = !ERTS_EMPTY_RUNQ(rq);
+ res = rq->halt_in_progress ?
+ !ERTS_EMPTY_RUNQ_PORTS(rq) : !ERTS_EMPTY_RUNQ(rq);
return res;
}
@@ -3583,6 +3658,7 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online)
rq->len = 0;
rq->wakeup_other = 0;
rq->wakeup_other_reds = 0;
+ rq->halt_in_progress = 0;
rq->procs.len = 0;
rq->procs.pending_exiters = NULL;
@@ -3777,6 +3853,9 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online)
ERTS_VERIFY_UNUSED_TEMP_ALLOC(NULL);
#endif
#endif
+
+ erts_smp_atomic32_init_relb(&erts_halt_progress, -1);
+ erts_halt_code = 0;
}
ErtsRunQueue *
@@ -6140,6 +6219,15 @@ Process *schedule(Process *p, int calls)
int actual_reds;
int reds;
+#ifdef USE_VM_PROBES
+ if (p != NULL && DTRACE_ENABLED(process_unscheduled)) {
+ DTRACE_CHARBUF(process_buf, DTRACE_TERM_BUF_SIZE);
+
+ dtrace_proc_str(p, process_buf);
+ DTRACE1(process_unscheduled, process_buf);
+ }
+#endif
+
if (ERTS_USE_MODIFIED_TIMING()) {
context_reds = ERTS_MODIFIED_TIMING_CONTEXT_REDS;
input_reductions = ERTS_MODIFIED_TIMING_INPUT_REDS;
@@ -6343,7 +6431,9 @@ Process *schedule(Process *p, int calls)
ASSERT(rq->len == rq->procs.len + rq->ports.info.len);
- if (rq->len == 0 && !rq->misc.start) {
+ if ((rq->len == 0 && !rq->misc.start)
+ || (rq->halt_in_progress
+ && rq->ports.info.len == 0 && !rq->misc.start)) {
#ifdef ERTS_SMP
@@ -6441,7 +6531,8 @@ Process *schedule(Process *p, int calls)
if (rq->ports.info.len) {
int have_outstanding_io;
have_outstanding_io = erts_port_task_execute(rq, &esdp->current_port);
- if (have_outstanding_io && fcalls > 2*input_reductions) {
+ if ((have_outstanding_io && fcalls > 2*input_reductions)
+ || rq->halt_in_progress) {
/*
* If we have performed more than 2*INPUT_REDUCTIONS since
* last call to erl_sys_schedule() and we still haven't
@@ -7175,6 +7266,10 @@ erl_create_process(Process* parent, /* Parent of process (default group leader).
p->seq_trace_lastcnt = 0;
p->seq_trace_clock = 0;
SEQ_TRACE_TOKEN(p) = NIL;
+#ifdef USE_VM_PROBES
+ DT_UTAG(p) = NIL;
+ DT_UTAG_FLAGS(p) = 0;
+#endif
p->parent = parent->id == ERTS_INVALID_PID ? NIL : parent->id;
#ifdef HYBRID
@@ -7307,6 +7402,16 @@ erl_create_process(Process* parent, /* Parent of process (default group leader).
VERBOSE(DEBUG_PROCESSES, ("Created a new process: %T\n",p->id));
+#ifdef USE_VM_PROBES
+ if (DTRACE_ENABLED(process_spawn)) {
+ DTRACE_CHARBUF(process_name, DTRACE_TERM_BUF_SIZE);
+ DTRACE_CHARBUF(mfa, DTRACE_TERM_BUF_SIZE);
+
+ dtrace_fun_decode(p, mod, func, arity, process_name, mfa);
+ DTRACE2(process_spawn, process_name, mfa);
+ }
+#endif
+
error:
erts_smp_proc_unlock(parent, ERTS_PROC_LOCKS_ALL_MINOR);
@@ -7758,7 +7863,11 @@ static ERTS_INLINE void
send_exit_message(Process *to, ErtsProcLocks *to_locksp,
Eterm exit_term, Uint term_size, Eterm token)
{
- if (token == NIL) {
+ if (token == NIL
+#ifdef USE_VM_PROBES
+ || token == am_have_dt_utag
+#endif
+ ) {
Eterm* hp;
Eterm mess;
ErlHeapFragment* bp;
@@ -7766,7 +7875,11 @@ send_exit_message(Process *to, ErtsProcLocks *to_locksp,
hp = erts_alloc_message_heap(term_size, &bp, &ohp, to, to_locksp);
mess = copy_struct(exit_term, term_size, &hp, ohp);
- erts_queue_message(to, to_locksp, bp, mess, NIL);
+ erts_queue_message(to, to_locksp, bp, mess, NIL
+#ifdef USE_VM_PROBES
+ , NIL
+#endif
+ );
} else {
ErlHeapFragment* bp;
Eterm* hp;
@@ -7782,7 +7895,11 @@ send_exit_message(Process *to, ErtsProcLocks *to_locksp,
/* the trace token must in this case be updated by the caller */
seq_trace_output(token, mess, SEQ_TRACE_SEND, to->id, NULL);
temp_token = copy_struct(token, sz_token, &hp, &bp->off_heap);
- erts_queue_message(to, to_locksp, bp, mess, temp_token);
+ erts_queue_message(to, to_locksp, bp, mess, temp_token
+#ifdef USE_VM_PROBES
+ , NIL
+#endif
+ );
}
}
@@ -7875,9 +7992,26 @@ send_exit_signal(Process *c_p, /* current process if and only
ASSERT(reason != THE_NON_VALUE);
+#ifdef USE_VM_PROBES
+ if(DTRACE_ENABLED(process_exit_signal) && is_pid(from)) {
+ DTRACE_CHARBUF(sender_str, DTRACE_TERM_BUF_SIZE);
+ DTRACE_CHARBUF(receiver_str, DTRACE_TERM_BUF_SIZE);
+ DTRACE_CHARBUF(reason_buf, DTRACE_TERM_BUF_SIZE);
+
+ dtrace_pid_str(from, sender_str);
+ dtrace_proc_str(rp, receiver_str);
+ erts_snprintf(reason_buf, sizeof(reason_buf) - 1, "%T", reason);
+ DTRACE3(process_exit_signal, sender_str, receiver_str, reason_buf);
+ }
+#endif
+
if (ERTS_PROC_IS_TRAPPING_EXITS(rp)
&& (reason != am_kill || (flags & ERTS_XSIG_FLG_IGN_KILL))) {
- if (is_not_nil(token) && token_update)
+ if (is_not_nil(token)
+#ifdef USE_VM_PROBES
+ && token != am_have_dt_utag
+#endif
+ && token_update)
seq_trace_update_send(token_update);
if (is_value(exit_tuple))
send_exit_message(rp, rp_locks, exit_tuple, exit_tuple_sz, token);
@@ -8301,7 +8435,18 @@ erts_do_exit_process(Process* p, Eterm reason)
p->arity = 0; /* No live registers */
p->fvalue = reason;
-
+
+#ifdef USE_VM_PROBES
+ if (DTRACE_ENABLED(process_exit)) {
+ DTRACE_CHARBUF(process_buf, DTRACE_TERM_BUF_SIZE);
+ DTRACE_CHARBUF(reason_buf, DTRACE_TERM_BUF_SIZE);
+
+ dtrace_proc_str(p, process_buf);
+ erts_snprintf(reason_buf, DTRACE_TERM_BUF_SIZE - 1, "%T", reason);
+ DTRACE2(process_exit, process_buf, reason_buf);
+ }
+#endif
+
#ifdef ERTS_SMP
ERTS_SMP_CHK_HAVE_ONLY_MAIN_PROC_LOCK(p);
/* By locking all locks (main lock is already locked) when going
@@ -9867,3 +10012,30 @@ debug_processes_assert_error(char* expr, char* file, int line)
/* *\
* End of the processes/0 BIF implementation. *
\* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+
+/*
+ * A nice system halt closing all open port goes as follows:
+ * 1) This function schedules the aux work ERTS_SSI_AUX_WORK_REAP_PORTS
+ * on all schedulers, then schedules itself out.
+ * 2) All shedulers detect this and set the flag halt_in_progress
+ * on their run queue. The last scheduler sets all non-closed ports
+ * ERTS_PORT_SFLG_HALT. Global atomic erts_halt_progress is used
+ * as refcount to determine which is last.
+ * 3) While the run ques has flag halt_in_progress no processes
+ * will be scheduled, only ports.
+ * 4) When the last port closes that scheduler calls erlang:halt/1.
+ * The same global atomic is used as refcount.
+ *
+ * A BIF that calls this should make sure to schedule out to never come back:
+ * erl_halt((int)(- code));
+ * ERTS_BIF_YIELD1(bif_export[BIF_erlang_halt_1], BIF_P, NIL);
+ */
+void erl_halt(int code)
+{
+ if (-1 == erts_smp_atomic32_cmpxchg_acqb(&erts_halt_progress,
+ erts_no_schedulers,
+ -1)) {
+ erts_halt_code = code;
+ notify_reap_ports_relb();
+ }
+}