erl_trace.c

OTP是开放电信平台的简称

    }
    ERTS_CHK_HAVE_ONLY_MAIN_PROC_LOCK(p->id);
    return result;
}

/* Sends trace message:
 *    {trace_ts, Pid, What, Msg, Timestamp}
 * or {trace, Pid, What, Msg}
 *
 * where 'What' must be atomic and 'Msg' is: 
 * [{heap_size, HeapSize}, {old_heap_size, OldHeapSize}, 
 *  {stack_size, StackSize}, {recent_size, RecentSize}, 
 *  {mbuf_size, MbufSize}]
 *
 * where 'HeapSize', 'OldHeapSize', 'StackSize', 'RecentSize and 'MbufSize'
 * are all small (atomic) integers.
 */
void
trace_gc(Process *p, Eterm what)
{
#define HEAP_WORDS_NEEDED 40
    ErlHeapFragment *bp = NULL;
    ErlOffHeap *off_heap;
#ifndef ERTS_SMP
    Process* tracer = NULL;	/* Initialized to eliminate compiler warning */
#endif
    Eterm* hp;
    Eterm msg = NIL;
    Eterm tuple;
    Uint size = HEAP_WORDS_NEEDED + TS_SIZE(p);
#ifdef DEBUG
    Eterm* limit;
#endif


#define CONS_PAIR(key, val) \
    tuple = TUPLE2(hp, key, val); hp += 3; \
    msg = CONS(hp, tuple, msg); hp += 2

    if (is_internal_port(p->tracer_proc)) {
	Eterm local_heap[HEAP_WORDS_NEEDED+5];
	ASSERT(size <= sizeof(local_heap) / sizeof(local_heap[0]));
	hp = local_heap;
    } else {
	ASSERT(is_internal_pid(p->tracer_proc)
	       && internal_pid_index(p->tracer_proc) < erts_max_processes);

#ifdef ERTS_SMP
	bp = new_message_buffer(size);
	hp = bp->mem;
	off_heap = &bp->off_heap;
#else
	tracer = process_tab[internal_pid_index(p->tracer_proc)];
	if (INVALID_PID(tracer, p->tracer_proc)
	    || (tracer->trace_flags & F_TRACER) == 0) {
	    p->trace_flags &= ~TRACEE_FLAGS;
	    p->tracer_proc = NIL;
	    return;
	}
	hp = erts_alloc_message_heap(size, &bp, &off_heap, tracer, 0);
#endif
    }
#undef HEAP_WORDS_NEEDED
#ifdef DEBUG
    limit = hp + size;
#endif

    CONS_PAIR(am_heap_size, make_small(HEAP_TOP(p) - HEAP_START(p)));
    CONS_PAIR(am_old_heap_size, make_small(OLD_HTOP(p) - OLD_HEAP(p)));
    CONS_PAIR(am_stack_size, make_small(STACK_START(p) - p->stop));
    CONS_PAIR(am_recent_size, make_small(HIGH_WATER(p) - HEAP_START(p)));
    CONS_PAIR(am_mbuf_size, make_small(MBUF_SIZE(p)));
    CONS_PAIR(am_heap_block_size, make_small(HEAP_SIZE(p)));
    CONS_PAIR(am_old_heap_block_size, make_small(OLD_HEAP(p)
						 ? OLD_HEND(p) - OLD_HEAP(p)
						 : 0));

    msg = TUPLE4(hp, am_trace, p->id/* Local pid */, what, msg);
    hp += 5;

    erts_smp_mtx_lock(&smq_mtx);

    if (p->trace_flags & F_TIMESTAMP) {
	hp = patch_ts(msg, hp);
    }
    ASSERT(hp == limit);
    if (is_internal_port(p->tracer_proc)) {
	send_to_port(p, msg, &p->tracer_proc, &p->trace_flags);
    } else {
#ifdef ERTS_SMP
	enqueue_sys_msg_unlocked(SYS_MSG_TYPE_TRACE,
				 p->id, p->tracer_proc, msg, bp);
#else
	erts_queue_message(tracer, 0, bp, msg, NIL);
#endif
    }
    erts_smp_mtx_unlock(&smq_mtx);
#undef CONS_PAIR
}



void
monitor_long_gc(Process *p, Uint time) {
    ErlHeapFragment *bp;
#ifndef ERTS_SMP
    ErlOffHeap *off_heap;
    Process *monitor_p;
#endif
    Uint hsz;
    Eterm *hp, timeout, tuple, list, msg;

#ifndef ERTS_SMP
    ASSERT(is_internal_pid(system_monitor)
	   && internal_pid_index(system_monitor) < erts_max_processes);
    monitor_p = process_tab[internal_pid_index(system_monitor)];
    if (INVALID_PID(monitor_p, system_monitor) || p == monitor_p) {
	return;
    }
#endif

#define CONS_PAIR(key, val) \
    tuple = TUPLE2(hp, (key), (val)); hp += 3; \
    list = CONS(hp, tuple, list); hp += 2

    hsz = 30;
    if (!IS_USMALL(0, time)) {
	hsz += BIG_UINT_HEAP_SIZE;
    }
#ifdef ERTS_SMP
    bp = new_message_buffer(hsz);
    hp = bp->mem;
#else
    hp = erts_alloc_message_heap(hsz, &bp, &off_heap, monitor_p, 0);
#endif
    if (IS_USMALL(0, time)) {
	timeout = make_small(time);
    } else {
	timeout = uint_to_big(time, hp);
	hp += BIG_UINT_HEAP_SIZE;
    }

    list = NIL;
    CONS_PAIR(am_heap_size, make_small(HEAP_TOP(p) - HEAP_START(p)));
    CONS_PAIR(am_stack_size, make_small(STACK_START(p) - p->stop));
    CONS_PAIR(am_mbuf_size, make_small(MBUF_SIZE(p)));
    CONS_PAIR(am_heap_block_size, make_small(HEAP_SIZE(p)));
    CONS_PAIR(am_timeout, timeout);
    msg = TUPLE4(hp, am_monitor, p->id/* Local pid */, am_long_gc, list); 
    hp += 5;

#ifdef ERTS_SMP
    enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->id, NIL, msg, bp);
#else
    erts_queue_message(monitor_p, 0, bp, msg, NIL);
#endif
#undef CONS_PAIR
}

void
monitor_large_heap(Process *p) {
    ErlHeapFragment *bp;
#ifndef ERTS_SMP
    ErlOffHeap *off_heap;
    Process *monitor_p;
#endif
    Eterm *hp, tuple, list, msg;
   
#ifndef ERTS_SMP 
    ASSERT(is_internal_pid(system_monitor)
	   && internal_pid_index(system_monitor) < erts_max_processes);
    monitor_p = process_tab[internal_pid_index(system_monitor)];
    if (INVALID_PID(monitor_p, system_monitor) || p == monitor_p) {
	return;
    }
#endif

#define CONS_PAIR(key, val) \
    tuple = TUPLE2(hp, (key), (val)); hp += 3; \
    list = CONS(hp, tuple, list); hp += 2


#ifdef ERTS_SMP
    bp = new_message_buffer(25);
    hp = bp->mem;
#else
    hp = erts_alloc_message_heap(25, &bp, &off_heap, monitor_p, 0);
#endif

    list = NIL;
    CONS_PAIR(am_heap_size, make_small(HEAP_TOP(p) - HEAP_START(p)));
    CONS_PAIR(am_stack_size, make_small(STACK_START(p) - p->stop));
    CONS_PAIR(am_mbuf_size, make_small(MBUF_SIZE(p)));
    CONS_PAIR(am_heap_block_size, make_small(HEAP_SIZE(p)));
    msg = TUPLE4(hp, am_monitor, p->id/* Local pid */, am_large_heap, list); 
    hp += 5;

#ifdef ERTS_SMP
    enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->id, NIL, msg, bp);
#else
    erts_queue_message(monitor_p, 0, bp, msg, NIL);
#endif
#undef CONS_PAIR
}

void
monitor_generic(Process *p, Eterm type, Eterm spec) {
    ErlHeapFragment *bp;
#ifndef ERTS_SMP
    ErlOffHeap *off_heap;
    Process *monitor_p;
#endif
    Eterm *hp, msg;

#ifndef ERTS_SMP
    ASSERT(is_internal_pid(system_monitor)
	   && internal_pid_index(system_monitor) < erts_max_processes);
    monitor_p = process_tab[internal_pid_index(system_monitor)];
    if (INVALID_PID(monitor_p, system_monitor) || p == monitor_p) {
	return;
    }
#endif

#ifdef ERTS_SMP
    bp = new_message_buffer(5);
    hp = bp->mem;
#else
    hp = erts_alloc_message_heap(5, &bp, &off_heap, monitor_p, 0);
#endif

    msg = TUPLE4(hp, am_monitor, p->id/* Local pid */, type, spec); 
    hp += 5;

#ifdef ERTS_SMP
    enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->id, NIL, msg, bp);
#else
    erts_queue_message(monitor_p, 0, bp, msg, NIL);
#endif

}


#ifdef ERTS_SMP

void
erts_check_my_tracer_proc(Process *p)
{
    if (is_internal_pid(p->tracer_proc)) {
	Process *tracer = erts_pid2proc(p, ERTS_PROC_LOCK_MAIN,
					p->tracer_proc, ERTS_PROC_LOCK_STATUS);
	int invalid_tracer = !tracer || !(tracer->trace_flags & F_TRACER);
	if (tracer)
	    erts_smp_proc_unlock(tracer, ERTS_PROC_LOCK_STATUS);
	if (invalid_tracer) {
	    erts_smp_proc_lock(p, ERTS_PROC_LOCKS_ALL_MINOR);
	    p->trace_flags &= ~TRACEE_FLAGS;
	    p->tracer_proc = NIL;
	    erts_smp_proc_unlock(p, ERTS_PROC_LOCKS_ALL_MINOR);
	}
    }
}


typedef struct ErtsSysMsgQ_ ErtsSysMsgQ;
struct  ErtsSysMsgQ_ {
    ErtsSysMsgQ *next;
    enum ErtsSysMsgType type;
    Eterm from;
    Eterm to;
    Eterm msg;
    ErlHeapFragment *bp;
};

static ErtsSysMsgQ *sys_message_queue;
static ErtsSysMsgQ *sys_message_queue_end;

static erts_tid_t sys_msg_dispatcher_tid;
static erts_cnd_t smq_cnd;

static int dispatcher_waiting;

ERTS_QUALLOC_IMPL(smq_element, ErtsSysMsgQ, 20, ERTS_ALC_T_SYS_MSG_Q)

static void
enqueue_sys_msg_unlocked(enum ErtsSysMsgType type,
			 Eterm from,
			 Eterm to,
			 Eterm msg,
			 ErlHeapFragment *bp)
{
    ErtsSysMsgQ *smqp;

    smqp	= smq_element_alloc();
    smqp->next	= NULL;
    smqp->type	= type;
    smqp->from	= from;
    smqp->to	= to;
    smqp->msg	= msg;
    smqp->bp	= bp;
    
    if (sys_message_queue_end) {
	ASSERT(sys_message_queue);
	sys_message_queue_end->next = smqp;
    }
    else {
	ASSERT(!sys_message_queue);
	sys_message_queue = smqp;
    }
    sys_message_queue_end = smqp;
    erts_smp_cnd_signal(&smq_cnd);
}

static void
enqueue_sys_msg(enum ErtsSysMsgType type,
		Eterm from,
		Eterm to,
		Eterm msg,
		ErlHeapFragment *bp)
{
    erts_smp_mtx_lock(&smq_mtx);
    enqueue_sys_msg_unlocked(type, from, to, msg, bp);
    erts_smp_mtx_unlock(&smq_mtx);
}

static void
prepare_for_block(void *unused)
{
    erts_smp_mtx_unlock(&smq_mtx);
}

static void
resume_after_block(void *unused)
{
    erts_smp_mtx_lock(&smq_mtx);
}

void
erts_queue_error_logger_message(Eterm from, Eterm msg, ErlHeapFragment *bp)
{
    enqueue_sys_msg(SYS_MSG_TYPE_ERRLGR, from, am_error_logger, msg, bp);
}

void
erts_send_sys_msg_proc(Eterm from, Eterm to, Eterm msg, ErlHeapFragment *bp)
{
    ASSERT(is_internal_pid(to));
    enqueue_sys_msg(SYS_MSG_TYPE_PROC_MSG, from, to, msg, bp);
}

#ifdef DEBUG_PRINTOUTS
static void
print_msg_type(ErtsSysMsgQ *smqp)
{
    switch (smqp->type) {
    case SYS_MSG_TYPE_TRACE:
	erts_fprintf(stderr, "TRACE ");
	break;
    case SYS_MSG_TYPE_SEQTRACE:
	erts_fprintf(stderr, "SEQTRACE ");
	break;
    case SYS_MSG_TYPE_SYSMON:
	erts_fprintf(stderr, "SYSMON ");
	break;
    case SYS_MSG_TYPE_ERRLGR:
	erts_fprintf(stderr, "ERRLGR ");
	break;
    case SYS_MSG_TYPE_PROC_MSG:
	erts_fprintf(stderr, "PROC_MSG ");
	break;
    default:
	erts_fprintf(stderr, "??? ");
	break;
    }
}
#endif

static void
sys_msg_disp_failure(ErtsSysMsgQ *smqp, Eterm receiver)
{
    switch (smqp->type) {
    case SYS_MSG_TYPE_TRACE:
	/* Invalid tracer_proc's are removed when processes
	   are scheduled in. */
	break;
    case SYS_MSG_TYPE_SEQTRACE:
	/* Reset seq_tracer if it hasn't changed */
	erts_smp_mtx_lock(&sys_trace_mtx);
	if (system_seq_tracer == receiver)
	    system_seq_tracer = am_false;
	erts_smp_mtx_unlock(&sys_trace_mtx);
	break;
    case SYS_MSG_TYPE_SYSMON:
	if (receiver == NIL
	    && !erts_system_monitor_long_gc
	    && !erts_system_monitor_large_heap
	    && !erts_system_monitor_flags.busy_port
	    && !erts_system_monitor_flags.busy_dist_port)
	    break; /* Everything is disabled */
	erts_smp_block_system(ERTS_BS_FLG_ALLOW_GC);
	if (system_monitor == receiver || receiver == NIL)
	    erts_system_monitor_clear(NULL);
	erts_smp_release_system();
	break;
    case SYS_MSG_TYPE_ERRLGR: {
	char *no_elgger = "(no error logger present)";
	Eterm *tp;
	Eterm tag;
	if (is_not_tuple(smqp->msg)) {
	unexpected_elmsg:
	    erts_fprintf(stderr,
			 "%s unexpected error logger message: %T\n",
			 no_elgger,
			 smqp->msg);
	}

	tp = tuple_val(smqp->msg);
	if (arityval(tp[0]) != 2)
	    goto unexpected_elmsg;
	if (is_not_tuple(tp[2]))
	    goto unexpected_elmsg;
	tp = tuple_val(tp[2]);
	if (arityval(tp[0]) != 3)
	    goto unexpected_elmsg;
	tag = tp[1];
	if (is_not_tuple(tp[3]))
	    goto unexpected_elmsg;
	tp = tuple_val(tp[3]);
	if (arityval(tp[0]) != 3)
	    goto unexpected_elmsg;
	if (is_not_list(tp[3]))
	    goto unexpected_elmsg;
	erts_fprintf(stderr, "%s %T: %T\n",
		     no_elgger, tag, CAR(list_val(tp[3])));
	break;
    }
    case SYS_MSG_TYPE_PROC_MSG:
	break;
    default:
	ASSERT(0);
    }
}

static void *
sys_msg_dispatcher_func(void *unused)
{
    ErtsSysMsgQ *local_sys_message_queue = NULL;

#ifdef ERTS_ENABLE_LOCK_CHECK
    erts_lc_set_thread_name("system message dispatcher");
#endif

    erts_register_blockable_thread();
    erts_smp_activity_begin(ERTS_ACTIVITY_IO, NULL, NULL, NULL);

    while (1) {
	ErtsSysMsgQ *smqp;

	ERTS_SMP_LC_ASSERT(!ERTS_LC_IS_BLOCKING);

	erts_smp_mtx_lock(&smq_mtx);

	/* Free previously used queue ... */
	while (local_sys_message_queue) {
	    smqp = local_sys_message_queue;
	    local_sys_message_queue = smqp->next;
	    smq_element_free(smqp);
	}

	/* Fetch current trace message queue ... */
	erts_smp_activity_change(ERTS_ACTIVITY_IO,
				 ERTS_ACTIVITY_WAIT,
				 prepare_for_block,
				 resume_after_block,
				 NULL);
	dispatcher_waiting = 1;
	while (!sys_message_queue)
	    erts_smp_cnd_wait(&smq_cnd, &smq_mtx);
	dispatcher_waiting = 0;
	erts_smp_activity_change(ERTS_ACTIVITY_WAIT,
				 ERTS_ACTIVITY_IO,
				 prepare_for_block,
				 resume_after_block,
				 NULL);

	local_sys_message_queue = sys_message_queue;
	sys_message_queue = NULL;
	sys_message_queue_end = NULL;

	erts_smp_mtx_unlock(&smq_mtx);

	/* Send trace messages ... */

	ASSERT(local_sys_message_queue);

	for (smqp = local_sys_message_queue; smqp; smqp = smqp->next) {
	    Eterm receiver;
	    Uint32 proc_locks = ERTS_PROC_LOCKS_MSG_SEND;
	    Process *proc = NULL;
	    Port *port = NULL;

#ifdef DEBUG_PRINTOUTS
	    print_msg_type(smqp);
#endif
	    switch (smqp->type) {
	    case SYS_MSG_TYPE_TRACE:
	    case SYS_MSG_TYPE_PROC_MSG:
		receiver = smqp->to;
		break;
	    case SYS_MSG_TYPE_SEQTRACE:
		receiver = erts_get_system_seq_tracer();
		break;
	    case SYS_MSG_TYPE_SYSMON:
		receiver = erts_get_system_monitor();
		if (smqp->from == receiver) {
#ifdef DEBUG_PRINTOUTS
		    erts_fprintf(stderr, "MSG=%T to %T... ",
				 smqp->msg, receiver);
#endif
		    goto drop_sys_msg;
		}
		break;
	    case SYS_MSG_TYPE