path: root/erts/emulator/beam/erl_message.c

/*
 * %CopyrightBegin%
 *
 * Copyright Ericsson AB 1997-2011. All Rights Reserved.
 *
 * The contents of this file are subject to the Erlang Public License,
 * Version 1.1, (the "License"); you may not use this file except in
 * compliance with the License. You should have received a copy of the
 * Erlang Public License along with this software. If not, it can be
 * retrieved online at http://www.erlang.org/.
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 *
 * %CopyrightEnd%
 */
/*
 * Message passing primitives.
 */

#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#include "sys.h"
#include "erl_vm.h"
#include "global.h"
#include "erl_message.h"
#include "erl_process.h"
#include "erl_nmgc.h"
#include "erl_binary.h"
#include "dtrace-wrapper.h"

ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(message,
				 ErlMessage,
				 ERL_MESSAGE_BUF_SZ,
				 ERTS_ALC_T_MSG_REF)

#if defined(DEBUG) && 0
#define HARD_DEBUG
#else
#undef HARD_DEBUG
#endif




static ERTS_INLINE int in_heapfrag(const Eterm* ptr, const ErlHeapFragment *bp)
{
    return ((unsigned)(ptr - bp->mem) < bp->used_size);
}


void
init_message(void)
{
    init_message_alloc();
}

void
free_message(ErlMessage* mp)
{
    message_free(mp);
}

/* Allocate message buffer (size in words) */
ErlHeapFragment*
new_message_buffer(Uint size)
{
    ErlHeapFragment* bp;
    bp = (ErlHeapFragment*) ERTS_HEAP_ALLOC(ERTS_ALC_T_HEAP_FRAG,
					    ERTS_HEAP_FRAG_SIZE(size));
    ERTS_INIT_HEAP_FRAG(bp, size);
    return bp;
}

ErlHeapFragment*
erts_resize_message_buffer(ErlHeapFragment *bp, Uint size,
			   Eterm *brefs, Uint brefs_size)
{
#ifdef DEBUG
    int i;
#endif
#ifdef HARD_DEBUG
    ErlHeapFragment *dbg_bp;
    Eterm *dbg_brefs;
    Uint dbg_size;
    Uint dbg_tot_size;
    Eterm *dbg_hp;
#endif
    ErlHeapFragment* nbp;

    /* ToDo: Make use of 'used_size' to avoid realloc
	when shrinking just a few words */

#ifdef DEBUG
    {
	Uint off_sz = size < bp->used_size ? size : bp->used_size;
	for (i = 0; i < brefs_size; i++) {
	    Eterm *ptr;
	    if (is_immed(brefs[i]))
		continue;
	    ptr = ptr_val(brefs[i]);
	    ASSERT(&bp->mem[0] <= ptr && ptr < &bp->mem[0] + off_sz);

	}
    }
#endif

    if (size == bp->used_size)
	return bp;

#ifdef HARD_DEBUG
    dbg_brefs = erts_alloc(ERTS_ALC_T_UNDEF, sizeof(Eterm *)*brefs_size);
    dbg_bp = new_message_buffer(bp->used_size);
    dbg_hp = dbg_bp->mem;
    dbg_tot_size = 0;
    for (i = 0; i < brefs_size; i++) {
	dbg_size = size_object(brefs[i]);
	dbg_tot_size += dbg_size;
	dbg_brefs[i] = copy_struct(brefs[i], dbg_size, &dbg_hp,
				   &dbg_bp->off_heap);
    }
    ASSERT(dbg_tot_size == (size < bp->used_size ? size : bp->used_size));
#endif

    nbp = (ErlHeapFragment*) ERTS_HEAP_REALLOC(ERTS_ALC_T_HEAP_FRAG,
					       (void *) bp,
					       ERTS_HEAP_FRAG_SIZE(bp->alloc_size),
					       ERTS_HEAP_FRAG_SIZE(size));
    if (bp != nbp) {
	Uint off_sz = size < nbp->used_size ? size : nbp->used_size;
	Eterm *sp = &bp->mem[0];
	Eterm *ep = sp + off_sz;
	Sint offs = &nbp->mem[0] - sp;
	erts_offset_off_heap(&nbp->off_heap, offs, sp, ep);
	erts_offset_heap(&nbp->mem[0], off_sz, offs, sp, ep);
	if (brefs && brefs_size)
	    erts_offset_heap_ptr(brefs, brefs_size, offs, sp, ep);
#ifdef DEBUG
	for (i = 0; i < brefs_size; i++) {
	    Eterm *ptr;
	    if (is_immed(brefs[i]))
		continue;
	    ptr = ptr_val(brefs[i]);
	    ASSERT(&nbp->mem[0] <= ptr && ptr < &nbp->mem[0] + off_sz);
	}
#endif
    }
    nbp->alloc_size = size;
    nbp->used_size = size;

#ifdef HARD_DEBUG
    for (i = 0; i < brefs_size; i++)
	ASSERT(eq(dbg_brefs[i], brefs[i]));
    free_message_buffer(dbg_bp);
    erts_free(ERTS_ALC_T_UNDEF, dbg_brefs);
#endif

    return nbp;
}


void
erts_cleanup_offheap(ErlOffHeap *offheap)
{
    union erl_off_heap_ptr u;

    for (u.hdr = offheap->first; u.hdr; u.hdr = u.hdr->next) {
	switch (thing_subtag(u.hdr->thing_word)) {
	case REFC_BINARY_SUBTAG:
	    if (erts_refc_dectest(&u.pb->val->refc, 0) == 0) {
		erts_bin_free(u.pb->val);
	    }
	    break;
	case FUN_SUBTAG:
	    if (erts_refc_dectest(&u.fun->fe->refc, 0) == 0) {
		erts_erase_fun_entry(u.fun->fe);		    
	    }
	    break;
	default:
	    ASSERT(is_external_header(u.hdr->thing_word));
	    erts_deref_node_entry(u.ext->node);
	    break;
	}
    }
}

void
free_message_buffer(ErlHeapFragment* bp)
{
    ASSERT(bp != NULL);
    do {
	ErlHeapFragment* next_bp = bp->next;

	erts_cleanup_offheap(&bp->off_heap);
	ERTS_HEAP_FREE(ERTS_ALC_T_HEAP_FRAG, (void *) bp,
		       ERTS_HEAP_FRAG_SIZE(bp->size));	
	bp = next_bp;
    }while (bp != NULL);
}

static ERTS_INLINE void
link_mbuf_to_proc(Process *proc, ErlHeapFragment *bp)
{
    if (bp) {
	/* Link the message buffer */
	bp->next = MBUF(proc);
	MBUF(proc) = bp;
	MBUF_SIZE(proc) += bp->used_size;
	FLAGS(proc) |= F_FORCE_GC;

	/* Move any off_heap's into the process */
	if (bp->off_heap.first != NULL) {
	    struct erl_off_heap_header** next_p = &bp->off_heap.first;
	    while (*next_p != NULL) {
		next_p = &((*next_p)->next);
	    }
	    *next_p = MSO(proc).first;
	    MSO(proc).first = bp->off_heap.first;
	    bp->off_heap.first = NULL;
	    OH_OVERHEAD(&(MSO(proc)), bp->off_heap.overhead);
	}
    }
}

Eterm
erts_msg_distext2heap(Process *pp,
		      ErtsProcLocks *plcksp,
		      ErlHeapFragment **bpp,
		      Eterm *tokenp,
		      ErtsDistExternal *dist_extp)
{
    Eterm msg;
    Uint tok_sz = 0;
    Eterm *hp = NULL;
    Eterm *hp_end = NULL;
    ErlOffHeap *ohp;
    Sint sz;

    *bpp = NULL;
    sz = erts_decode_dist_ext_size(dist_extp);
    if (sz < 0)
	goto decode_error;
    if (is_not_nil(*tokenp)) {
	ErlHeapFragment *heap_frag = erts_dist_ext_trailer(dist_extp);
	tok_sz = heap_frag->used_size;
	sz += tok_sz;
    }
    if (pp)
	hp = erts_alloc_message_heap(sz, bpp, &ohp, pp, plcksp);
    else {
	*bpp = new_message_buffer(sz);
	hp = (*bpp)->mem;
	ohp = &(*bpp)->off_heap;
    }
    hp_end = hp + sz;
    msg = erts_decode_dist_ext(&hp, ohp, dist_extp);
    if (is_non_value(msg))
	goto decode_error;
    if (is_not_nil(*tokenp)) {
	ErlHeapFragment *heap_frag = erts_dist_ext_trailer(dist_extp);
	*tokenp = copy_struct(*tokenp, tok_sz, &hp, ohp);
	erts_cleanup_offheap(&heap_frag->off_heap);
    }
    erts_free_dist_ext_copy(dist_extp);
    if (hp_end != hp) {
	if (!(*bpp)) {
	    HRelease(pp, hp_end, hp);
	}
	else {
	    Uint final_size = hp - &(*bpp)->mem[0];
	    Eterm brefs[2] = {msg, *tokenp};
	    ASSERT(sz - (hp_end - hp) == final_size);
	    *bpp = erts_resize_message_buffer(*bpp, final_size, &brefs[0], 2);
	    msg = brefs[0];
	    *tokenp = brefs[1];
	}
    }
    return msg;

 decode_error:
    if (is_not_nil(*tokenp)) {
	ErlHeapFragment *heap_frag = erts_dist_ext_trailer(dist_extp);
	erts_cleanup_offheap(&heap_frag->off_heap);
    }
    erts_free_dist_ext_copy(dist_extp);
    if (*bpp) {
	free_message_buffer(*bpp);
	*bpp = NULL;
    }    
    else if (hp) {
	HRelease(pp, hp_end, hp);
    }
    return THE_NON_VALUE;
 }

static ERTS_INLINE void
notify_new_message(Process *receiver)
{
    ERTS_SMP_LC_ASSERT(ERTS_PROC_LOCK_STATUS
		       & erts_proc_lc_my_proc_locks(receiver));

    ACTIVATE(receiver);

    switch (receiver->status) {
    case P_GARBING:
	switch (receiver->gcstatus) {
	case P_SUSPENDED:
	    goto suspended;
	case P_WAITING:
	    goto waiting;
	default:
	    break;
	}
	break;
    case P_SUSPENDED:
    suspended:
	receiver->rstatus = P_RUNABLE;
	break;
    case P_WAITING:
    waiting:
	erts_add_to_runq(receiver);
	break;
    default:
	break;
    }
}

void
erts_queue_dist_message(Process *rcvr,
			ErtsProcLocks *rcvr_locks,
			ErtsDistExternal *dist_ext,
			Eterm token)
{
    ErlMessage* mp;
    ERTS_DECLARE_DUMMY(Sint tok_label) = 0;
    ERTS_DECLARE_DUMMY(Sint tok_lastcnt) = 0;
    ERTS_DECLARE_DUMMY(Sint tok_serial) = 0;
#ifdef ERTS_SMP
    ErtsProcLocks need_locks;
#endif

    ERTS_SMP_LC_ASSERT(*rcvr_locks == erts_proc_lc_my_proc_locks(rcvr));

    mp = message_alloc();

#ifdef ERTS_SMP
    need_locks = ~(*rcvr_locks) & (ERTS_PROC_LOCK_MSGQ|ERTS_PROC_LOCK_STATUS);
    if (need_locks) {
	*rcvr_locks |= need_locks;
	if (erts_smp_proc_trylock(rcvr, need_locks) == EBUSY) {
	    if (need_locks == ERTS_PROC_LOCK_MSGQ) {
		erts_smp_proc_unlock(rcvr, ERTS_PROC_LOCK_STATUS);
		need_locks = (ERTS_PROC_LOCK_MSGQ
			      | ERTS_PROC_LOCK_STATUS);
	    }
	    erts_smp_proc_lock(rcvr, need_locks);
	}
    }

    if (rcvr->is_exiting || ERTS_PROC_PENDING_EXIT(rcvr)) {
	/* Drop message if receiver is exiting or has a pending exit ... */
	if (is_not_nil(token)) {
	    ErlHeapFragment *heap_frag;
	    heap_frag = erts_dist_ext_trailer(mp->data.dist_ext);
	    erts_cleanup_offheap(&heap_frag->off_heap);
	}
	erts_free_dist_ext_copy(dist_ext);
	message_free(mp);
    }
    else
#endif
    if (IS_TRACED_FL(rcvr, F_TRACE_RECEIVE)) {
	/* Ahh... need to decode it in order to trace it... */
	ErlHeapFragment *mbuf;
	Eterm msg;
	message_free(mp);
	msg = erts_msg_distext2heap(rcvr, rcvr_locks, &mbuf, &token, dist_ext);
	if (is_value(msg))
            if (DTRACE_ENABLED(message_queued)) {
                DTRACE_CHARBUF(receiver_name, DTRACE_TERM_BUF_SIZE);

                dtrace_proc_str(rcvr, receiver_name);
                if (token != NIL) {
                    tok_label = signed_val(SEQ_TRACE_T_LABEL(token));
                    tok_lastcnt = signed_val(SEQ_TRACE_T_LASTCNT(token));
                    tok_serial = signed_val(SEQ_TRACE_T_SERIAL(token));
                }
                DTRACE6(message_queued,
                        receiver_name, size_object(msg), rcvr->msg.len,
                        tok_label, tok_lastcnt, tok_serial);
            }
	    erts_queue_message(rcvr, rcvr_locks, mbuf, msg, token
#ifdef USE_VM_PROBES
			       , NIL
#endif
			       );
    }
    else {
	/* Enqueue message on external format */

	ERL_MESSAGE_TERM(mp) = THE_NON_VALUE;
#ifdef USE_VM_PROBES
	ERL_MESSAGE_DT_UTAG(mp) = NIL;
	if (token == am_have_dt_utag) {
	    ERL_MESSAGE_TOKEN(mp) = NIL;
	} else {
#endif
	    ERL_MESSAGE_TOKEN(mp) = token;
#ifdef USE_VM_PROBES
	}
#endif
	mp->next = NULL;

        if (DTRACE_ENABLED(message_queued)) {
            DTRACE_CHARBUF(receiver_name, DTRACE_TERM_BUF_SIZE);

            dtrace_proc_str(rcvr, receiver_name);
            if (token != NIL) {
                tok_label = signed_val(SEQ_TRACE_T_LABEL(token));
                tok_lastcnt = signed_val(SEQ_TRACE_T_LASTCNT(token));
                tok_serial = signed_val(SEQ_TRACE_T_SERIAL(token));
            }
            /*
             * TODO: We don't know the real size of the external message here.
             *       -1 will appear to a D script as 4294967295.
             */
            DTRACE6(message_queued, receiver_name, -1, rcvr->msg.len + 1,
                    tok_label, tok_lastcnt, tok_serial);
        }
	mp->data.dist_ext = dist_ext;
	LINK_MESSAGE(rcvr, mp);

	notify_new_message(rcvr);
    }
}

/* Add a message last in message queue */
void
erts_queue_message(Process* receiver,
		   ErtsProcLocks *receiver_locks,
		   ErlHeapFragment* bp,
		   Eterm message,
		   Eterm seq_trace_token
#ifdef USE_VM_PROBES
		   , Eterm dt_utag
#endif
)
{
    ErlMessage* mp;
#ifdef ERTS_SMP
    ErtsProcLocks need_locks;
#else
    ASSERT(bp != NULL || receiver->mbuf == NULL);
#endif

    ERTS_SMP_LC_ASSERT(*receiver_locks == erts_proc_lc_my_proc_locks(receiver));

    mp = message_alloc();

#ifdef ERTS_SMP
    need_locks = ~(*receiver_locks) & (ERTS_PROC_LOCK_MSGQ
				       | ERTS_PROC_LOCK_STATUS);
    if (need_locks) {
	*receiver_locks |= need_locks;
	if (erts_smp_proc_trylock(receiver, need_locks) == EBUSY) {
	    if (need_locks == ERTS_PROC_LOCK_MSGQ) {
		erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_STATUS);
		need_locks = (ERTS_PROC_LOCK_MSGQ
			      | ERTS_PROC_LOCK_STATUS);
	    }
	    erts_smp_proc_lock(receiver, need_locks);
	}
    }

    if (receiver->is_exiting || ERTS_PROC_PENDING_EXIT(receiver)) {
	/* Drop message if receiver is exiting or has a pending
	 * exit ...
	 */
	if (bp)
	    free_message_buffer(bp);
	message_free(mp);
	return;
    }
#endif

    ERL_MESSAGE_TERM(mp) = message;
    ERL_MESSAGE_TOKEN(mp) = seq_trace_token;
#ifdef USE_VM_PROBES
    ERL_MESSAGE_DT_UTAG(mp) = dt_utag;
#endif
    mp->next = NULL;
    mp->data.heap_frag = bp;

#ifdef ERTS_SMP
    if (*receiver_locks & ERTS_PROC_LOCK_MAIN) {
	/*
	 * We move 'in queue' to 'private queue' and place
	 * message at the end of 'private queue' in order
	 * to ensure that the 'in queue' doesn't contain
	 * references into the heap. By ensuring this,
	 * we don't need to include the 'in queue' in
	 * the root set when garbage collecting.
	 */
	ERTS_SMP_MSGQ_MV_INQ2PRIVQ(receiver);
	LINK_MESSAGE_PRIVQ(receiver, mp);
    }
    else {
	LINK_MESSAGE(receiver, mp);
    }
#else
    LINK_MESSAGE(receiver, mp);
#endif

    if (DTRACE_ENABLED(message_queued)) {
        DTRACE_CHARBUF(receiver_name, DTRACE_TERM_BUF_SIZE);
        ERTS_DECLARE_DUMMY(Sint tok_label) = 0;
        ERTS_DECLARE_DUMMY(Sint tok_lastcnt) = 0;
        ERTS_DECLARE_DUMMY(Sint tok_serial) = 0;

        dtrace_proc_str(receiver, receiver_name);
        if (seq_trace_token != NIL && is_tuple(seq_trace_token)) {
            tok_label = signed_val(SEQ_TRACE_T_LABEL(seq_trace_token));
            tok_lastcnt = signed_val(SEQ_TRACE_T_LASTCNT(seq_trace_token));
            tok_serial = signed_val(SEQ_TRACE_T_SERIAL(seq_trace_token));
        }
        DTRACE6(message_queued,
                receiver_name, size_object(message), receiver->msg.len,
                tok_label, tok_lastcnt, tok_serial);
    }

    notify_new_message(receiver);

    if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
	trace_receive(receiver, message);
    }

#ifndef ERTS_SMP
    ERTS_HOLE_CHECK(receiver);
#endif
}

void
erts_link_mbuf_to_proc(struct process *proc, ErlHeapFragment *bp)
{
    Eterm* htop = HEAP_TOP(proc);

    link_mbuf_to_proc(proc, bp);
    if (htop < HEAP_LIMIT(proc)) {
	*htop = make_pos_bignum_header(HEAP_LIMIT(proc)-htop-1);
	HEAP_TOP(proc) = HEAP_LIMIT(proc);
    }
}

/*
 * Moves content of message buffer attached to a message into a heap.
 * The message buffer is deallocated.
 */
void
erts_move_msg_mbuf_to_heap(Eterm** hpp, ErlOffHeap* off_heap, ErlMessage *msg)
{
    struct erl_off_heap_header* oh;
    Eterm term, token, *fhp, *hp;
    Sint offs;
    Uint sz;
    ErlHeapFragment *bp;
#ifdef USE_VM_PROBES
    Eterm utag;
#endif

#ifdef HARD_DEBUG
    ProcBin *dbg_mso_start = off_heap->mso;
    ErlFunThing *dbg_fun_start = off_heap->funs;
    ExternalThing *dbg_external_start = off_heap->externals;
    Eterm dbg_term, dbg_token;
    ErlHeapFragment *dbg_bp;
    Uint *dbg_hp, *dbg_thp_start;
    Uint dbg_term_sz, dbg_token_sz;
#ifdef USE_VM_PROBES
    Eterm dbg_utag;
    Uint dbg_utag_sz;
#endif
#endif

    bp = msg->data.heap_frag;
    term = ERL_MESSAGE_TERM(msg);
    token = ERL_MESSAGE_TOKEN(msg);
#ifdef USE_VM_PROBES
    utag = ERL_MESSAGE_DT_UTAG(msg);
#endif
    if (!bp) {
#ifdef USE_VM_PROBES
	ASSERT(is_immed(term) && is_immed(token) && is_immed(utag));
#else
	ASSERT(is_immed(term) && is_immed(token));
#endif
	return;
    }

#ifdef HARD_DEBUG
    dbg_term_sz = size_object(term);
    dbg_token_sz = size_object(token);
    dbg_bp = new_message_buffer(dbg_term_sz + dbg_token_sz);
#ifdef USE_VM_PROBES
    dbg_utag_sz = size_object(utag);
    dbg_bp = new_message_buffer(dbg_term_sz + dbg_token_sz + dbg_utag_sz );
#endif
    /*ASSERT(dbg_term_sz + dbg_token_sz == erts_msg_used_frag_sz(msg));
      Copied size may be smaller due to removed SubBins's or garbage.
      Copied size may be larger due to duplicated shared terms.
    */
    dbg_hp = dbg_bp->mem;
    dbg_term = copy_struct(term, dbg_term_sz, &dbg_hp, &dbg_bp->off_heap);
    dbg_token = copy_struct(token, dbg_token_sz, &dbg_hp, &dbg_bp->off_heap);
#ifdef USE_VM_PROBES
    dbg_utag = copy_struct(utag, dbg_utag_sz, &dbg_hp, &dbg_bp->off_heap);
#endif
   dbg_thp_start = *hpp;
#endif

    if (bp->next != NULL) {
	move_multi_frags(hpp, off_heap, bp, msg->m, 
#ifdef USE_VM_PROBES
			 3
#else
			 2
#endif
			 );
	goto copy_done;
    }

    OH_OVERHEAD(off_heap, bp->off_heap.overhead);
    sz = bp->used_size;

    ASSERT(is_immed(term) || in_heapfrag(ptr_val(term),bp));
    ASSERT(is_immed(token) || in_heapfrag(ptr_val(token),bp));

    fhp = bp->mem;
    hp = *hpp;
    offs = hp - fhp;

    oh = NULL;
    while (sz--) {
	Uint cpy_sz;
	Eterm val = *fhp++;

	switch (primary_tag(val)) {
	case TAG_PRIMARY_IMMED1:
	    *hp++ = val;
	    break;
	case TAG_PRIMARY_LIST:
	case TAG_PRIMARY_BOXED:
	    ASSERT(in_heapfrag(ptr_val(val), bp));
	    *hp++ = offset_ptr(val, offs);
	    break;
	case TAG_PRIMARY_HEADER:
	    *hp++ = val;
	    switch (val & _HEADER_SUBTAG_MASK) {
	    case ARITYVAL_SUBTAG:
		break;
	    case REFC_BINARY_SUBTAG:
	    case FUN_SUBTAG:
	    case EXTERNAL_PID_SUBTAG:
	    case EXTERNAL_PORT_SUBTAG:
	    case EXTERNAL_REF_SUBTAG:
		oh = (struct erl_off_heap_header*) (hp-1);
		cpy_sz = thing_arityval(val);
		goto cpy_words;
	    default:
		cpy_sz = header_arity(val);

	    cpy_words:
		ASSERT(sz >= cpy_sz);
		sz -= cpy_sz;
		while (cpy_sz >= 8) {
		    cpy_sz -= 8;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		    *hp++ = *fhp++;
		}
		switch (cpy_sz) {
		case 7: *hp++ = *fhp++;
		case 6: *hp++ = *fhp++;
		case 5: *hp++ = *fhp++;
		case 4: *hp++ = *fhp++;
		case 3: *hp++ = *fhp++;
		case 2: *hp++ = *fhp++;
		case 1: *hp++ = *fhp++;
		default: break;
		}
		if (oh) {
		    /* Add to offheap list */
		    oh->next = off_heap->first;
		    off_heap->first = oh;
		    ASSERT(*hpp <= (Eterm*)oh);
		    ASSERT(hp > (Eterm*)oh);
		    oh = NULL;
		}
		break;
	    }
	    break;
	}
    }

    ASSERT(bp->used_size == hp - *hpp);
    *hpp = hp;

    if (is_not_immed(token)) {
	ASSERT(in_heapfrag(ptr_val(token), bp));
	ERL_MESSAGE_TOKEN(msg) = offset_ptr(token, offs);
#ifdef HARD_DEBUG
	ASSERT(dbg_thp_start <= ptr_val(ERL_MESSAGE_TOKEN(msg)));
	ASSERT(hp > ptr_val(ERL_MESSAGE_TOKEN(msg)));
#endif
    }

    if (is_not_immed(term)) {
	ASSERT(in_heapfrag(ptr_val(term),bp));
	ERL_MESSAGE_TERM(msg) = offset_ptr(term, offs);
#ifdef HARD_DEBUG
	ASSERT(dbg_thp_start <= ptr_val(ERL_MESSAGE_TERM(msg)));
	ASSERT(hp > ptr_val(ERL_MESSAGE_TERM(msg)));
#endif
    }
#ifdef USE_VM_PROBES
    if (is_not_immed(utag)) {
	ASSERT(in_heapfrag(ptr_val(utag), bp));
	ERL_MESSAGE_DT_UTAG(msg) = offset_ptr(utag, offs);
#ifdef HARD_DEBUG
	ASSERT(dbg_thp_start <= ptr_val(ERL_MESSAGE_DT_UTAG(msg)));
	ASSERT(hp > ptr_val(ERL_MESSAGE_DT_UTAG(msg)));
#endif
    }
#endif

copy_done:

#ifdef HARD_DEBUG
    {
	int i, j;
	ErlHeapFragment* frag;
	{
	    ProcBin *mso = off_heap->mso;
	    i = j = 0;
	    while (mso != dbg_mso_start) {
		mso = mso->next;
		i++;
	    }
	    for (frag=bp; frag; frag=frag->next) {
		mso = frag->off_heap.mso;
		while (mso) {
		    mso = mso->next;
		    j++;
		}
	    }
	    ASSERT(i == j);
	}
	{
	    ErlFunThing *fun = off_heap->funs;
	    i = j = 0;
	    while (fun != dbg_fun_start) {
		fun = fun->next;
		i++;
	    }
	    for (frag=bp; frag; frag=frag->next) {
		fun = frag->off_heap.funs;
		while (fun) {
		    fun = fun->next;
		    j++;
		}
	    }
	    ASSERT(i == j);
	}
	{
	    ExternalThing *external = off_heap->externals;
	    i = j = 0;
	    while (external != dbg_external_start) {
		external = external->next;
		i++;
	    }
	    for (frag=bp; frag; frag=frag->next) {
		external = frag->off_heap.externals;
		while (external) {
		    external = external->next;
		    j++;
		}
	    }
	    ASSERT(i == j);
	}
    }
#endif
	    

    bp->off_heap.first = NULL;
    free_message_buffer(bp);
    msg->data.heap_frag = NULL;

#ifdef HARD_DEBUG
    ASSERT(eq(ERL_MESSAGE_TERM(msg), dbg_term));
    ASSERT(eq(ERL_MESSAGE_TOKEN(msg), dbg_token));
#ifdef USE_VM_PROBES
    ASSERT(eq(ERL_MESSAGE_DT_UTAG(msg), dbg_utag));
#endif
    free_message_buffer(dbg_bp);
#endif

}


Uint
erts_msg_attached_data_size_aux(ErlMessage *msg)
{
    Sint sz;
    ASSERT(is_non_value(ERL_MESSAGE_TERM(msg)));
    ASSERT(msg->data.dist_ext);
    ASSERT(msg->data.dist_ext->heap_size < 0);

    sz = erts_decode_dist_ext_size(msg->data.dist_ext);
    if (sz < 0) {
	/* Bad external; remove it */
	if (is_not_nil(ERL_MESSAGE_TOKEN(msg))) {
	    ErlHeapFragment *heap_frag;
	    heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
	    erts_cleanup_offheap(&heap_frag->off_heap);
	}
	erts_free_dist_ext_copy(msg->data.dist_ext);
	msg->data.dist_ext = NULL;
	return 0;
    }

    msg->data.dist_ext->heap_size = sz;
    if (is_not_nil(msg->m[1])) {
	ErlHeapFragment *heap_frag;
	heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
	sz += heap_frag->used_size;
    }
    return sz;
}

void
erts_move_msg_attached_data_to_heap(Eterm **hpp, ErlOffHeap *ohp, ErlMessage *msg)
{
    if (is_value(ERL_MESSAGE_TERM(msg)))
	erts_move_msg_mbuf_to_heap(hpp, ohp, msg);
    else if (msg->data.dist_ext) {
	ASSERT(msg->data.dist_ext->heap_size >= 0);
	if (is_not_nil(ERL_MESSAGE_TOKEN(msg))) {
	    ErlHeapFragment *heap_frag;
	    heap_frag = erts_dist_ext_trailer(msg->data.dist_ext);
	    ERL_MESSAGE_TOKEN(msg) = copy_struct(ERL_MESSAGE_TOKEN(msg),
						 heap_frag->used_size,
						 hpp,
						 ohp);
	    erts_cleanup_offheap(&heap_frag->off_heap);
	}
	ERL_MESSAGE_TERM(msg) = erts_decode_dist_ext(hpp,
						     ohp,
						     msg->data.dist_ext);
	erts_free_dist_ext_copy(msg->data.dist_ext);
	msg->data.dist_ext = NULL;
    }
    /* else: bad external detected when calculating size */
}

/*
 * Send a local message when sender & receiver processes are known.
 */

void
erts_send_message(Process* sender,
		  Process* receiver,
		  ErtsProcLocks *receiver_locks,
		  Eterm message,
		  unsigned flags)
{
    Uint msize;
    ErlHeapFragment* bp = NULL;
    Eterm token = NIL;
    DTRACE_CHARBUF(sender_name, 64);
    DTRACE_CHARBUF(receiver_name, 64);
    ERTS_DECLARE_DUMMY(Sint tok_label) = 0;
    ERTS_DECLARE_DUMMY(Sint tok_lastcnt) = 0;
    ERTS_DECLARE_DUMMY(Sint tok_serial) = 0;

    BM_STOP_TIMER(system);
    BM_MESSAGE(message,sender,receiver);
    BM_START_TIMER(send);

    *sender_name = *receiver_name = '\0';
    if (DTRACE_ENABLED(message_send)) {
        erts_snprintf(sender_name, sizeof(sender_name), "%T", sender->id);
        erts_snprintf(receiver_name, sizeof(receiver_name), "%T", receiver->id);
    }
    if (SEQ_TRACE_TOKEN(sender) != NIL && !(flags & ERTS_SND_FLG_NO_SEQ_TRACE)) {
        Eterm* hp;
	Eterm stoken = SEQ_TRACE_TOKEN(sender);
	Uint seq_trace_size = 0;
#ifdef USE_VM_PROBES
	Uint dt_utag_size = 0;
	Eterm utag = NIL;
#endif

	BM_SWAP_TIMER(send,size);
	msize = size_object(message);
	BM_SWAP_TIMER(size,send);

#ifdef USE_VM_PROBES
	if (stoken != am_have_dt_utag) {
#endif

	    seq_trace_update_send(sender);
	    seq_trace_output(stoken, message, SEQ_TRACE_SEND, 
			     receiver->id, sender);
	    seq_trace_size = 6; /* TUPLE5 */
#ifdef USE_VM_PROBES
	}
	if (DT_UTAG_FLAGS(sender) & DT_UTAG_SPREADING) {
	    dt_utag_size = size_object(DT_UTAG(sender));
	} else if (stoken == am_have_dt_utag ) {
	    stoken = NIL;
	}
#endif

	bp = new_message_buffer(msize + seq_trace_size 
#ifdef USE_VM_PROBES
				+ dt_utag_size
#endif
				);
	hp = bp->mem;

        BM_SWAP_TIMER(send,copy);
	token = copy_struct(stoken,
			    seq_trace_size,
			    &hp,
			    &bp->off_heap);

	message = copy_struct(message, msize, &hp, &bp->off_heap);
#ifdef USE_VM_PROBES
	if (DT_UTAG_FLAGS(sender) & DT_UTAG_SPREADING) {
	    utag = copy_struct(DT_UTAG(sender), dt_utag_size, &hp, &bp->off_heap);
#ifdef DTRACE_TAG_HARDDEBUG
	    erts_fprintf(stderr,
			 "Dtrace -> (%T) Spreading tag (%T) with "
			 "message %T!\r\n",sender->id, utag, message);
#endif
	}
#endif
        BM_MESSAGE_COPIED(msize);
        BM_SWAP_TIMER(copy,send);

        if (DTRACE_ENABLED(message_send)) {
	    if (stoken != NIL) {
		tok_label = signed_val(SEQ_TRACE_T_LABEL(stoken));
		tok_lastcnt = signed_val(SEQ_TRACE_T_LASTCNT(stoken));
		tok_serial = signed_val(SEQ_TRACE_T_SERIAL(stoken));
	    }
	    DTRACE6(message_send, sender_name, receiver_name,
		    msize, tok_label, tok_lastcnt, tok_serial);
        }
        erts_queue_message(receiver,
			   receiver_locks,
			   bp,
			   message,
			   token
#ifdef USE_VM_PROBES
			   , utag
#endif
			   );
        BM_SWAP_TIMER(send,system);
#ifdef HYBRID
    } else {
        ErlMessage* mp = message_alloc();
        BM_SWAP_TIMER(send,copy);
#ifdef INCREMENTAL
        /* TODO: During GC activate processes if the message relies in
         * the fromspace and the sender is active. During major
         * collections add the message to the gray stack if it relies
         * in the old generation and the sender is active and the
         * receiver is inactive.

        if (!IS_CONST(message) && (ma_gc_flags & GC_CYCLE) &&
            (ptr_val(message) >= inc_fromspc &&
            ptr_val(message) < inc_fromend) && INC_IS_ACTIVE(sender))
            INC_ACTIVATE(receiver);
        else if (!IS_CONST(message) && (ma_gc_flags & GC_CYCLE) &&
            (ptr_val(message) >= global_old_heap &&
            ptr_val(message) < global_old_hend) &&
            INC_IS_ACTIVE(sender) && !INC_IS_ACTIVE(receiver))
            Mark message in blackmap and add it to the gray stack
        */

         if (!IS_CONST(message))
            INC_ACTIVATE(receiver);
#endif
        LAZY_COPY(sender,message);
        BM_SWAP_TIMER(copy,send);
        DTRACE6(message_send, sender_name, receiver_name,
                size_object(message)msize, tok_label, tok_lastcnt, tok_serial);
        ERL_MESSAGE_TERM(mp) = message;
        ERL_MESSAGE_TOKEN(mp) = NIL;
#ifdef USE_VM_PROBES
	ERL_MESSAGE_DT_UTAG(mp) = NIL;
#endif
        mp->next = NULL;
	LINK_MESSAGE(receiver, mp);
        ACTIVATE(receiver);

        if (receiver->status == P_WAITING) {
            erts_add_to_runq(receiver);
        } else if (receiver->status == P_SUSPENDED) {
            receiver->rstatus = P_RUNABLE;
        }
        if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
            trace_receive(receiver, message);
        }

        BM_SWAP_TIMER(send,system);
        return;
#else
    } else if (sender == receiver) {
	/* Drop message if receiver has a pending exit ... */
#ifdef ERTS_SMP
	ErtsProcLocks need_locks = (~(*receiver_locks)
				    & (ERTS_PROC_LOCK_MSGQ
				       | ERTS_PROC_LOCK_STATUS));
	if (need_locks) {
	    *receiver_locks |= need_locks;
	    if (erts_smp_proc_trylock(receiver, need_locks) == EBUSY) {
		if (need_locks == ERTS_PROC_LOCK_MSGQ) {
		    erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_STATUS);
		    need_locks = ERTS_PROC_LOCK_MSGQ|ERTS_PROC_LOCK_STATUS;
		}
		erts_smp_proc_lock(receiver, need_locks);
	    }
	}
	if (!ERTS_PROC_PENDING_EXIT(receiver))
#endif
	{
	    ErlMessage* mp = message_alloc();

            DTRACE6(message_send, sender_name, receiver_name,
                    size_object(message), tok_label, tok_lastcnt, tok_serial);
	    mp->data.attached = NULL;
	    ERL_MESSAGE_TERM(mp) = message;
	    ERL_MESSAGE_TOKEN(mp) = NIL;
#ifdef USE_VM_PROBES
	    ERL_MESSAGE_DT_UTAG(mp) = NIL;
#endif
	    mp->next = NULL;
	    /*
	     * We move 'in queue' to 'private queue' and place
	     * message at the end of 'private queue' in order
	     * to ensure that the 'in queue' doesn't contain
	     * references into the heap. By ensuring this,
	     * we don't need to include the 'in queue' in
	     * the root set when garbage collecting.
	     */
	    
	    ERTS_SMP_MSGQ_MV_INQ2PRIVQ(receiver);
	    LINK_MESSAGE_PRIVQ(receiver, mp);

	    if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
		trace_receive(receiver, message);
	    }
	}
        BM_SWAP_TIMER(send,system);
	return;
    } else {
#ifdef ERTS_SMP
	ErlOffHeap *ohp;
        Eterm *hp;
	BM_SWAP_TIMER(send,size);
	msize = size_object(message);
	BM_SWAP_TIMER(size,send);
	hp = erts_alloc_message_heap(msize,&bp,&ohp,receiver,receiver_locks);
	BM_SWAP_TIMER(send,copy);
	message = copy_struct(message, msize, &hp, ohp);
	BM_MESSAGE_COPIED(msz);
	BM_SWAP_TIMER(copy,send);
        DTRACE6(message_send, sender_name, receiver_name,
                msize, tok_label, tok_lastcnt, tok_serial);
	erts_queue_message(receiver, receiver_locks, bp, message, token
#ifdef USE_VM_PROBES
			   , NIL
#endif
			   );
        BM_SWAP_TIMER(send,system);
#else
	ErlMessage* mp = message_alloc();
        Eterm *hp;
        BM_SWAP_TIMER(send,size);
	msize = size_object(message);
        BM_SWAP_TIMER(size,send);
	
	if (receiver->stop - receiver->htop <= msize) {
            BM_SWAP_TIMER(send,system);
	    erts_garbage_collect(receiver, msize, receiver->arg_reg, receiver->arity);
            BM_SWAP_TIMER(system,send);
	}
	hp = receiver->htop;
	receiver->htop = hp + msize;
        BM_SWAP_TIMER(send,copy);
	message = copy_struct(message, msize, &hp, &receiver->off_heap);
	BM_MESSAGE_COPIED(msize);
        BM_SWAP_TIMER(copy,send);
        DTRACE6(message_send, sender_name, receiver_name,
                (uint32_t)msize, tok_label, tok_lastcnt, tok_serial);
	ERL_MESSAGE_TERM(mp) = message;
	ERL_MESSAGE_TOKEN(mp) = NIL;
#ifdef USE_VM_PROBES
	ERL_MESSAGE_DT_UTAG(mp) = NIL;
#endif
	mp->next = NULL;
	mp->data.attached = NULL;
	LINK_MESSAGE(receiver, mp);

	if (receiver->status == P_WAITING) {
	    erts_add_to_runq(receiver);
	} else if (receiver->status == P_SUSPENDED) {
	    receiver->rstatus = P_RUNABLE;
	}
	if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
	    trace_receive(receiver, message);
	}
        BM_SWAP_TIMER(send,system);
#endif /* #ifndef ERTS_SMP */
	return;
#endif /* HYBRID */
    }
}

/*
 * This function delivers an EXIT message to a process
 * which is trapping EXITs.
 */

void
erts_deliver_exit_message(Eterm from, Process *to, ErtsProcLocks *to_locksp,
			  Eterm reason, Eterm token)
{
    Eterm mess;
    Eterm save;
    Eterm from_copy;
    Uint sz_reason;
    Uint sz_token;
    Uint sz_from;
    Eterm* hp;
    Eterm temptoken;
    ErlHeapFragment* bp = NULL;

    if (token != NIL
#ifdef USE_VM_PROBES
	&& token != am_have_dt_utag
#endif
	) {

	ASSERT(is_tuple(token));
	sz_reason = size_object(reason);
	sz_token = size_object(token);
	sz_from = size_object(from);
	bp = new_message_buffer(sz_reason + sz_from + sz_token + 4);
	hp = bp->mem;
	mess = copy_struct(reason, sz_reason, &hp, &bp->off_heap);
	from_copy = copy_struct(from, sz_from, &hp, &bp->off_heap);
	save = TUPLE3(hp, am_EXIT, from_copy, mess);
	hp += 4;
	/* the trace token must in this case be updated by the caller */
	seq_trace_output(token, save, SEQ_TRACE_SEND, to->id, NULL);
	temptoken = copy_struct(token, sz_token, &hp, &bp->off_heap);
	erts_queue_message(to, to_locksp, bp, save, temptoken
#ifdef USE_VM_PROBES
			   , NIL
#endif
			   );
    } else {
	ErlOffHeap *ohp;
	sz_reason = size_object(reason);
	sz_from = IS_CONST(from) ? 0 : size_object(from);

	hp = erts_alloc_message_heap(sz_reason+sz_from+4,
				     &bp,
				     &ohp,
				     to,
				     to_locksp);

	mess = copy_struct(reason, sz_reason, &hp, ohp);
	from_copy = (IS_CONST(from)
		     ? from
		     : copy_struct(from, sz_from, &hp, ohp));
	save = TUPLE3(hp, am_EXIT, from_copy, mess);
	erts_queue_message(to, to_locksp, bp, save, NIL
#ifdef USE_VM_PROBES
			   , NIL
#endif
			   );
    }
}