erl_bif_timer.c

OTP是开放电信平台的简称

			}
			hp = HAlloc(rp, wrap_size);
#ifdef ERTS_SMP
		    }
		    else
#endif
#endif /* #ifndef HEAP_FRAG_ELIM_TEST */
#if defined(ERTS_SMP) || defined(HEAP_FRAG_ELIM_TEST)
			if (!bp) {
			ErlOffHeap *ohp;
			ASSERT(is_immed(message));
			hp = erts_alloc_message_heap(wrap_size,
						     &bp,
						     &ohp,
						     rp,
						     &rp_locks);
		    }
		    else {
			Eterm old_size = bp->size;
			bp = erts_resize_message_buffer(bp, old_size + wrap_size,
							&message, 1);
			hp = &bp->mem[0] + old_size;
		    }
#endif
		    write_ref_thing(hp,
				    btm->ref_numbers[0],
				    btm->ref_numbers[1],
				    btm->ref_numbers[2]);
		    ref = make_internal_ref(hp);
		    hp += REF_THING_SIZE;
		    message = TUPLE3(hp, am_timeout, ref, message);
		}

		erts_queue_message(rp, rp_locks, bp, message, NIL);
	    }
	    erts_smp_proc_unlock(rp, rp_locks);
	}
    }

    erts_smp_btm_rwunlock();

    bif_timer_cleanup(btm);
}

static Eterm
setup_bif_timer(Uint32 xflags,
		Process *c_p,
		Eterm time,
		Eterm receiver,
		Eterm message)
{
    Process *rp;
    ErtsBifTimer* btm;
    Uint timeout;
    Eterm ref;
    Uint32 *ref_numbers;
    
    if (!term_to_Uint(time, &timeout))
	return THE_NON_VALUE;
#ifdef ARCH_64
    if ((timeout >> 32) != 0)
	return THE_NON_VALUE;
#endif
    if (is_not_internal_pid(receiver) && is_not_atom(receiver))
	return THE_NON_VALUE;

    ref = erts_make_ref(c_p);

    if (is_atom(receiver))
	rp = NULL;
    else {
	rp = erts_pid2proc(c_p, ERTS_PROC_LOCK_MAIN,
			   receiver, ERTS_PROC_LOCK_MSGQ);
	if (!rp)
	    return ref;
    }

    ERTS_PROC_MORE_MEM(sizeof(ErtsBifTimer));
    if (timeout < ERTS_ALC_MIN_LONG_LIVED_TIME) {
	if (timeout < 1000) {
	    btm = btm_pre_alloc();
	    if (!btm)
		goto sl_timer_alloc;
	    btm->flags = 0;
	}
	else {
	sl_timer_alloc:
	    btm = (ErtsBifTimer *) erts_alloc(ERTS_ALC_T_SL_BIF_TIMER,
					      sizeof(ErtsBifTimer));
	    btm->flags = BTM_FLG_SL_TIMER;
	}
    }
    else {
	btm = (ErtsBifTimer *) erts_alloc(ERTS_ALC_T_LL_BIF_TIMER,
					 sizeof(ErtsBifTimer));
	btm->flags = 0;
    }

    if (rp) {
	link_proc(rp, btm);
	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_MSGQ);
    }
    else {
	ASSERT(is_atom(receiver));
	btm->receiver.name = receiver;
	btm->flags |= BTM_FLG_BYNAME;
    }

    btm->flags |= xflags;

    ref_numbers = internal_ref_numbers(ref);
    ASSERT(internal_ref_no_of_numbers(ref) == 3);
#if ERTS_REF_NUMBERS != 3
#error "ERTS_REF_NUMBERS changed. Update me..."
#endif
    btm->ref_numbers[0] = ref_numbers[0];
    btm->ref_numbers[1] = ref_numbers[1];
    btm->ref_numbers[2] = ref_numbers[2];

    ASSERT(eq_ref_numbers(btm->ref_numbers, ERTS_REF_NUMBERS,
			  ref_numbers, ERTS_REF_NUMBERS));

    if (is_immed(message)) {
	btm->bp = NULL;
	btm->message = message;
    }
    else {
	ErlHeapFragment* bp;
	Eterm* hp;
	Uint size;

	size = size_object(message);
	btm->bp = bp = new_message_buffer(size);
	hp = bp->mem;
	btm->message = copy_struct(message, size, &hp, &bp->off_heap);
    }

    tab_insert(btm);
    ASSERT(btm == tab_find(ref));
    btm->tm.active = 0; /* MUST be initalized */
    erl_set_timer(&btm->tm,
		  (ErlTimeoutProc) bif_timer_timeout,
		  (ErlCancelProc) bif_timer_cleanup,
		  (void *) btm,
		  timeout);
    return ref;
}

/* send_after(Time, Pid, Message) -> Ref */
BIF_RETTYPE send_after_3(BIF_ALIST_3)
{
    Eterm res;

    if (erts_smp_safe_btm_rwlock(BIF_P, ERTS_PROC_LOCK_MAIN))
	ERTS_BIF_EXITED(BIF_P);

    res = setup_bif_timer(0, BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3);

    erts_smp_btm_rwunlock();

    if (is_non_value(res)) {
	BIF_ERROR(BIF_P, BADARG);
    }
    else {
	ASSERT(is_internal_ref(res));
	BIF_RET(res);
    }
}

/* start_timer(Time, Pid, Message) -> Ref */
BIF_RETTYPE start_timer_3(BIF_ALIST_3)
{
    Eterm res;

    if (erts_smp_safe_btm_rwlock(BIF_P, ERTS_PROC_LOCK_MAIN))
	ERTS_BIF_EXITED(BIF_P);

    res = setup_bif_timer(BTM_FLG_WRAP, BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3);

    erts_smp_btm_rwunlock();

    if (is_non_value(res)) {
	BIF_ERROR(BIF_P, BADARG);
    }
    else {
	ASSERT(is_internal_ref(res));
	BIF_RET(res);
    }
}

/* cancel_timer(Ref) -> false | RemainingTime */
BIF_RETTYPE cancel_timer_1(BIF_ALIST_1)
{
    Eterm res;
    ErtsBifTimer *btm;

    if (is_not_internal_ref(BIF_ARG_1)) {
	if (is_ref(BIF_ARG_1)) {
	    BIF_RET(am_false);
	}
	BIF_ERROR(BIF_P, BADARG);
    }

    if (erts_smp_safe_btm_rwlock(BIF_P, ERTS_PROC_LOCK_MAIN))
	ERTS_BIF_EXITED(BIF_P);

    btm = tab_find(BIF_ARG_1);
    if (!btm || btm->flags & BTM_FLG_CANCELED) {
	erts_smp_btm_rwunlock();
	res = am_false;
    }
    else {
	Uint left = time_left(&btm->tm);
	if (!(btm->flags & BTM_FLG_BYNAME)) {
	    erts_smp_proc_lock(btm->receiver.proc.ess, ERTS_PROC_LOCK_MSGQ);
	    unlink_proc(btm);
	    erts_smp_proc_unlock(btm->receiver.proc.ess, ERTS_PROC_LOCK_MSGQ);
	}
	tab_remove(btm);
	ASSERT(!tab_find(BIF_ARG_1));
	erl_cancel_timer(&btm->tm);
	erts_smp_btm_rwunlock();
	res = erts_make_integer(left, BIF_P);
    }

    BIF_RET(res);
}

/* read_timer(Ref) -> false | RemainingTime */
BIF_RETTYPE read_timer_1(BIF_ALIST_1)
{
    Eterm res;
    ErtsBifTimer *btm;

    if (is_not_internal_ref(BIF_ARG_1)) {
	if (is_ref(BIF_ARG_1)) {
	    BIF_RET(am_false);
	}
	BIF_ERROR(BIF_P, BADARG);
    }

    if (erts_smp_safe_btm_rlock(BIF_P, ERTS_PROC_LOCK_MAIN))
	ERTS_BIF_EXITED(BIF_P);

    btm = tab_find(BIF_ARG_1);
    if (!btm || btm->flags & BTM_FLG_CANCELED) {
	res = am_false;
    }
    else {
	Uint left = time_left(&btm->tm);
	res = erts_make_integer(left, BIF_P);
    }

    erts_smp_btm_runlock();

    BIF_RET(res);
}

void
erts_print_bif_timer_info(int to, void *to_arg)
{
    int i;
    int lock = !ERTS_IS_CRASH_DUMPING;

    if (lock)
	erts_smp_btm_rlock();

    for (i = 0; i < TIMER_HASH_VEC_SZ; i++) {
	ErtsBifTimer *btm;
	for (btm = bif_timer_tab[i]; btm; btm = btm->tab.next) {
	    Eterm receiver = (btm->flags & BTM_FLG_BYNAME
			      ? btm->receiver.name
			      : btm->receiver.proc.ess->id);
	    erts_print(to, to_arg, "=timer:%T\n", receiver);
	    erts_print(to, to_arg, "Message: %T\n", btm->message);
	    erts_print(to, to_arg, "Time left: %d ms\n", time_left(&btm->tm));
	}
    }

    if (lock)
	erts_smp_btm_runlock();
}


void
erts_cancel_bif_timers(struct process *p, Uint32 plocks)
{
    ErtsBifTimer *btm;

    if (erts_smp_btm_tryrwlock() == EBUSY) {
	erts_smp_proc_unlock(p, plocks);
	erts_smp_btm_rwlock();
	erts_smp_proc_lock(p, plocks);
    }

    btm = p->bif_timers;
    while (btm) {
	ErtsBifTimer *tmp_btm;
	ASSERT(!(btm->flags & BTM_FLG_CANCELED));
	tab_remove(btm);
	tmp_btm = btm;
	btm = btm->receiver.proc.next;
	erl_cancel_timer(&tmp_btm->tm);
    }

    p->bif_timers = NULL;

    erts_smp_btm_rwunlock();
}

void erts_bif_timer_init(void)
{
    int i;
    no_bif_timers = 0;
    init_btm_pre_alloc();
    erts_smp_btm_lock_init();
    bif_timer_tab = erts_alloc(ERTS_ALC_T_BIF_TIMER_TABLE,
			       sizeof(ErtsBifTimer *)*TIMER_HASH_VEC_SZ);
    for (i = 0; i < TIMER_HASH_VEC_SZ; ++i)
	bif_timer_tab[i] = NULL;
}

Uint
erts_bif_timer_memory_size(void)
{
    Uint res;
    int lock = !ERTS_IS_CRASH_DUMPING;

    if (lock)
	erts_smp_btm_rlock();

    res = (sizeof(ErtsBifTimer *)*TIMER_HASH_VEC_SZ
	   + no_bif_timers*sizeof(ErtsBifTimer));

    if (lock)
	erts_smp_btm_runlock();

    return res;
}


void
erts_bif_timer_foreach(void (*func)(Eterm, Eterm, ErlHeapFragment *, void *),
		       void *arg)
{
    int i;

    ERTS_SMP_LC_ASSERT(erts_smp_is_system_blocked(0));

    for (i = 0; i < TIMER_HASH_VEC_SZ; i++) {
	ErtsBifTimer *btm;
	for (btm = bif_timer_tab[i]; btm; btm = btm->tab.next) {
	    (*func)((btm->flags & BTM_FLG_BYNAME
		     ? btm->receiver.name
		     : btm->receiver.proc.ess->id),
		    btm->message,
		    btm->bp,
		    arg);
	}
    }
}