/* * %CopyrightBegin% * * Copyright Ericsson AB 1997-2018. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * * %CopyrightEnd% */ #ifndef __ERL_MESSAGE_H__ #define __ERL_MESSAGE_H__ #include "sys.h" #include "erl_vm.h" #define ERTS_PROC_SIG_QUEUE_TYPE_ONLY #include "erl_proc_sig_queue.h" #undef ERTS_PROC_SIG_QUEUE_TYPE_ONLY #ifdef DEBUG #define ERTS_MSG_COPY_WORDS_PER_REDUCTION 4 #else #define ERTS_MSG_COPY_WORDS_PER_REDUCTION 64 #endif struct proc_bin; struct external_thing_; typedef struct erl_mesg ErtsMessage; /* * This struct represents data that must be updated by structure copy, * but is stored outside of any heap. */ struct erl_off_heap_header { Eterm thing_word; Uint size; struct erl_off_heap_header* next; }; #define OH_OVERHEAD(oh, size) do { \ (oh)->overhead += size; \ } while(0) typedef struct erl_off_heap { struct erl_off_heap_header* first; Uint64 overhead; /* Administrative overhead (used to force GC). */ } ErlOffHeap; #define ERTS_INIT_OFF_HEAP(OHP) \ do { \ (OHP)->first = NULL; \ (OHP)->overhead = 0; \ } while (0) typedef struct { enum { FACTORY_CLOSED = 0, FACTORY_HALLOC, FACTORY_MESSAGE, FACTORY_HEAP_FRAGS, FACTORY_STATIC, FACTORY_TMP } mode; Process* p; Eterm* hp_start; Eterm* hp; Eterm* hp_end; ErtsMessage *message; struct erl_heap_fragment* heap_frags; struct erl_heap_fragment* heap_frags_saved; Uint heap_frags_saved_used; ErlOffHeap* off_heap; ErlOffHeap off_heap_saved; Uint32 alloc_type; } ErtsHeapFactory; void erts_factory_proc_init(ErtsHeapFactory*, Process*); void erts_factory_proc_prealloc_init(ErtsHeapFactory*, Process*, Sint size); void erts_factory_heap_frag_init(ErtsHeapFactory*, struct erl_heap_fragment*); ErtsMessage *erts_factory_message_create(ErtsHeapFactory *, Process *, ErtsProcLocks *, Uint sz); void erts_factory_selfcontained_message_init(ErtsHeapFactory*, ErtsMessage *, Eterm *); void erts_factory_static_init(ErtsHeapFactory*, Eterm* hp, Uint size, ErlOffHeap*); void erts_factory_tmp_init(ErtsHeapFactory*, Eterm* hp, Uint size, Uint32 atype); void erts_factory_dummy_init(ErtsHeapFactory*); ERTS_GLB_INLINE Eterm* erts_produce_heap(ErtsHeapFactory*, Uint need, Uint xtra); Eterm* erts_reserve_heap(ErtsHeapFactory*, Uint need); void erts_factory_close(ErtsHeapFactory*); void erts_factory_trim_and_close(ErtsHeapFactory*,Eterm *brefs, Uint brefs_size); void erts_factory_undo(ErtsHeapFactory*); void erts_reserve_heap__(ErtsHeapFactory*, Uint need, Uint xtra); /* internal */ #if ERTS_GLB_INLINE_INCL_FUNC_DEF ERTS_GLB_INLINE Eterm * erts_produce_heap(ErtsHeapFactory* factory, Uint need, Uint xtra) { Eterm* res; ASSERT((unsigned int)factory->mode > (unsigned int)FACTORY_CLOSED); if (factory->hp + need > factory->hp_end) { erts_reserve_heap__(factory, need, xtra); } res = factory->hp; factory->hp += need; INIT_HEAP_MEM(res, need); return res; } #endif /* ERTS_GLB_INLINE_INCL_FUNC_DEF */ #ifdef CHECK_FOR_HOLES # define ERTS_FACTORY_HOLE_CHECK(f) do { \ /*if ((f)->p) erts_check_for_holes((f)->p);*/ \ } while (0) #else # define ERTS_FACTORY_HOLE_CHECK(p) #endif #include "external.h" #include "erl_process.h" #define ERTS_INVALID_HFRAG_PTR ((ErlHeapFragment *) ~((UWord) 7)) /* * This struct represents a heap fragment, which is used when there * isn't sufficient room in the process heap and we can't do a GC. */ typedef struct erl_heap_fragment ErlHeapFragment; struct erl_heap_fragment { ErlHeapFragment* next; /* Next heap fragment */ ErlOffHeap off_heap; /* Offset heap data. */ Uint alloc_size; /* Size in words of mem */ Uint used_size; /* With terms to be moved to heap by GC */ Eterm mem[1]; /* Data */ }; /* m[0] = message, m[1] = seq trace token */ #define ERL_MESSAGE_REF_ARRAY_SZ 3 #define ERL_MESSAGE_TERM(mp) ((mp)->m[0]) #define ERL_MESSAGE_TOKEN(mp) ((mp)->m[1]) #define ERL_MESSAGE_FROM(mp) ((mp)->m[2]) #ifdef USE_VM_PROBES /* m[2] = dynamic trace user tag */ #undef ERL_MESSAGE_REF_ARRAY_SZ #define ERL_MESSAGE_REF_ARRAY_SZ 4 #define ERL_MESSAGE_DT_UTAG(mp) ((mp)->m[3]) #else #endif #ifdef USE_VM_PROBES #define have_no_seqtrace(T) ((T) == NIL || (T) == am_have_dt_utag) #else #define have_no_seqtrace(T) ((T) == NIL) #endif #define have_seqtrace(T) (!have_no_seqtrace(T)) #define ERL_MESSAGE_REF_FIELDS__ \ ErtsMessage *next; /* Next message */ \ union { \ ErlHeapFragment *heap_frag; \ void *attached; \ } data; \ Eterm m[ERL_MESSAGE_REF_ARRAY_SZ] typedef struct erl_msg_ref__ { ERL_MESSAGE_REF_FIELDS__; } ErtsMessageRef; struct erl_mesg { ERL_MESSAGE_REF_FIELDS__; ErlHeapFragment hfrag; }; /* * The ErtsMessage struct is only one special type * of signal. All signal structs have a common * begining and can be differentiated by looking * at the ErtsSignal 'common.tag' field. * * - An ordinary message will have a value * - A distribution message that has not been * decoded yet will have the non-value. * - Other signals will have an external pid * header tag. In order to differentiate * between those signals one needs to look * at the arity part of the header (see * erts_proc_sig_queue.h). */ #define ERTS_SIG_IS_NON_MSG_TAG(Tag) \ (is_external_pid_header((Tag))) #define ERTS_SIG_IS_NON_MSG(Sig) \ ERTS_SIG_IS_NON_MSG_TAG(((ErtsSignal *) (Sig))->common.tag) #define ERTS_SIG_IS_INTERNAL_MSG_TAG(Tag) \ (!is_header((Tag))) #define ERTS_SIG_IS_INTERNAL_MSG(Sig) \ ERTS_SIG_IS_INTERNAL_MSG_TAG(((ErtsSignal *) (Sig))->common.tag) #define ERTS_SIG_IS_EXTERNAL_MSG_TAG(Tag) \ ((Tag) == THE_NON_VALUE) #define ERTS_SIG_IS_EXTERNAL_MSG(Sig) \ ERTS_SIG_IS_EXTERNAL_MSG_TAG(((ErtsSignal *) (Sig))->common.tag) #define ERTS_SIG_IS_MSG_TAG(Tag) \ (!ERTS_SIG_IS_NON_MSG_TAG(Tag)) #define ERTS_SIG_IS_MSG(Sig) \ ERTS_SIG_IS_MSG_TAG(((ErtsSignal *) (Sig))->common.tag) typedef union { ErtsSignalCommon common; ErtsMessageRef msg; } ErtsSignal; typedef struct { /* pointers to next pointers pointing to... */ ErtsMessage **next; /* ... next (non-message) signal */ ErtsMessage **last; /* ... last (non-message) signal */ } ErtsMsgQNMSigs; /* Size of default message buffer (erl_message.c) */ #define ERL_MESSAGE_BUF_SZ 500 typedef struct { /* * ** The signal queues private to a process. ** * * These are: * - an inner queue which only consists of * message signals * - a middle queue which contains a mixture * of message and non-message signals * * When the process isn't processing signals in * erts_proc_sig_handle_incoming(): * - the message queue corresponds to the inner * queue. Messages in the middle queue (and * in the outer queue) are in transit and * have NOT been received yet! * * When the process is processing signals in * erts_proc_sig_handle_incoming(): * - the message queue corresponds to the inner * queue plus the head of the middle queue up * to the signal currently being processed. * Any messages further back in the middle queue * (and in the outer queue) are still in transit * and have NOT been received yet! * * In the general case the 'len' field of this * structure does NOT correspond to the message * queue length. When the process is inspected * via process info it does however correspond * to the message queue length, but this is a * special case! * * When no process-info request is in transit to * the process the 'len' field corresponds to * the total amount of messages in inner and * middle queues (which does NOT correspond to * the message queue length). When process-info * requests are in transit to the process, the * usage of the 'len' field changes and is used * as an offset which even might be negative. */ /* inner queue */ ErtsMessage *first; ErtsMessage **last; /* point to the last next pointer */ ErtsMessage **save; /* middle queue */ ErtsMessage *cont; ErtsMessage **cont_last; ErtsMsgQNMSigs nmsigs; /* Common for inner and middle queue */ ErtsMessage **saved_last; /* saved last pointer */ Sint len; /* NOT message queue length (see above) */ } ErtsSignalPrivQueues; typedef struct { ErtsMessage* first; ErtsMessage** last; /* point to the last next pointer */ Sint len; /* number of messages in queue */ ErtsMsgQNMSigs nmsigs; #ifdef ERTS_PROC_SIG_HARD_DEBUG int may_contain_heap_terms; #endif } ErtsSignalInQueue; typedef struct erl_trace_message_queue__ { struct erl_trace_message_queue__ *next; /* point to the next receiver */ Eterm receiver; ErtsMessage* first; ErtsMessage** last; /* point to the last next pointer */ Sint len; /* queue length */ } ErlTraceMessageQueue; #define ERTS_RECV_MARK_SAVE(P) \ do { \ erts_proc_lock((P), ERTS_PROC_LOCK_MSGQ); \ erts_proc_sig_fetch((P)); \ erts_proc_unlock((P), ERTS_PROC_LOCK_MSGQ); \ if ((P)->sig_qs.cont) { \ (P)->sig_qs.saved_last = (P)->sig_qs.cont_last; \ (P)->flags |= F_DEFERRED_SAVED_LAST; \ } \ else { \ (P)->sig_qs.saved_last = (P)->sig_qs.last; \ (P)->flags &= ~F_DEFERRED_SAVED_LAST; \ } \ } while (0) #define ERTS_RECV_MARK_SET(P) \ do { \ if ((P)->sig_qs.saved_last) { \ if ((P)->flags & F_DEFERRED_SAVED_LAST) { \ /* Points to middle queue; use end of inner */ \ (P)->sig_qs.save = (P)->sig_qs.last; \ ASSERT(!PEEK_MESSAGE((P))); \ } \ else { \ /* Points to inner queue; safe to use */ \ (P)->sig_qs.save = (P)->sig_qs.saved_last; \ } \ } \ } while (0) #define ERTS_RECV_MARK_CLEAR(P) \ do { \ (P)->sig_qs.saved_last = NULL; \ (P)->flags &= ~F_DEFERRED_SAVED_LAST; \ } while (0) /* Get "current" message */ #define PEEK_MESSAGE(p) (*(p)->sig_qs.save) #ifdef USE_VM_PROBES #define LINK_MESSAGE_DTAG(mp, dt) ERL_MESSAGE_DT_UTAG(mp) = dt #else #define LINK_MESSAGE_DTAG(mp, dt) #endif #ifdef USE_VM_PROBES # define ERTS_MSG_RECV_TRACED(P) \ ((ERTS_TRACE_FLAGS((P)) & F_TRACE_RECEIVE) \ || DTRACE_ENABLED(message_queued)) #else # define ERTS_MSG_RECV_TRACED(P) \ (ERTS_TRACE_FLAGS((P)) & F_TRACE_RECEIVE) #endif /* Add one message last in message queue */ #define LINK_MESSAGE(p, msg) \ do { \ ASSERT(ERTS_SIG_IS_MSG(msg)); \ ERTS_HDBG_CHECK_SIGNAL_IN_QUEUE__((p), "before"); \ *(p)->sig_inq.last = (msg); \ (p)->sig_inq.last = &(msg)->next; \ (p)->sig_inq.len++; \ ERTS_HDBG_CHECK_SIGNAL_IN_QUEUE__((p), "before"); \ } while(0) /* Unlink current message */ #define UNLINK_MESSAGE(p,msgp) \ do { \ ErtsMessage *mp__ = (msgp)->next; \ ERTS_HDBG_CHECK_SIGNAL_PRIV_QUEUE__((p), 0, "before"); \ *(p)->sig_qs.save = mp__; \ (p)->sig_qs.len--; \ if (mp__ == NULL) \ (p)->sig_qs.last = (p)->sig_qs.save; \ ERTS_HDBG_CHECK_SIGNAL_PRIV_QUEUE__((p), 0, "after"); \ } while(0) /* * Reset message save point (after receive match). * Also invalidate the saved position since it may no * longer be safe to use. */ #define JOIN_MESSAGE(p) \ do { \ (p)->sig_qs.save = &(p)->sig_qs.first; \ ERTS_RECV_MARK_CLEAR((p)); \ } while(0) /* Save current message */ #define SAVE_MESSAGE(p) \ (p)->sig_qs.save = &(*(p)->sig_qs.save)->next #define ERTS_HEAP_FRAG_SIZE(DATA_WORDS) \ (sizeof(ErlHeapFragment) - sizeof(Eterm) + (DATA_WORDS)*sizeof(Eterm)) #define ERTS_INIT_HEAP_FRAG(HEAP_FRAG_P, USED_WORDS, DATA_WORDS) \ do { \ (HEAP_FRAG_P)->next = NULL; \ (HEAP_FRAG_P)->alloc_size = (DATA_WORDS); \ (HEAP_FRAG_P)->used_size = (USED_WORDS); \ (HEAP_FRAG_P)->off_heap.first = NULL; \ (HEAP_FRAG_P)->off_heap.overhead = 0; \ } while (0) #ifdef USE_VM_PROBES #define ERL_MESSAGE_DT_UTAG_INIT(MP) ERL_MESSAGE_DT_UTAG(MP) = NIL #else #define ERL_MESSAGE_DT_UTAG_INIT(MP) do{ } while (0) #endif #define ERTS_INIT_MESSAGE(MP) \ do { \ (MP)->next = NULL; \ ERL_MESSAGE_TERM(MP) = THE_NON_VALUE; \ ERL_MESSAGE_TOKEN(MP) = THE_NON_VALUE; \ ERL_MESSAGE_FROM(MP) = NIL; \ ERL_MESSAGE_DT_UTAG_INIT(MP); \ MP->data.attached = NULL; \ } while (0) void init_message(void); ErlHeapFragment* new_message_buffer(Uint); ErlHeapFragment* erts_resize_message_buffer(ErlHeapFragment *, Uint, Eterm *, Uint); void free_message_buffer(ErlHeapFragment *); void erts_queue_dist_message(Process*, ErtsProcLocks, ErtsDistExternal *, ErlHeapFragment *, Eterm, Eterm); void erts_queue_message(Process*, ErtsProcLocks,ErtsMessage*, Eterm, Eterm); void erts_queue_proc_message(Process* from,Process* to, ErtsProcLocks,ErtsMessage*, Eterm); void erts_queue_proc_messages(Process* from, Process* to, ErtsProcLocks, ErtsMessage*, ErtsMessage**, Uint); void erts_deliver_exit_message(Eterm, Process*, ErtsProcLocks *, Eterm, Eterm); void erts_send_message(Process*, Process*, ErtsProcLocks*, Eterm); void erts_link_mbuf_to_proc(Process *proc, ErlHeapFragment *bp); Uint erts_msg_attached_data_size_aux(ErtsMessage *msg); void erts_cleanup_offheap(ErlOffHeap *offheap); void erts_save_message_in_proc(Process *p, ErtsMessage *msg); Sint erts_move_messages_off_heap(Process *c_p); Sint erts_complete_off_heap_message_queue_change(Process *c_p); Eterm erts_change_message_queue_management(Process *c_p, Eterm new_state); void erts_cleanup_messages(ErtsMessage *mp); void *erts_alloc_message_ref(void); void erts_free_message_ref(void *); #define ERTS_SMALL_FIX_MSG_SZ 10 #define ERTS_MEDIUM_FIX_MSG_SZ 20 #define ERTS_LARGE_FIX_MSG_SZ 30 void *erts_alloc_small_message(void); void erts_free_small_message(void *mp); typedef struct { ErtsMessage m; Eterm data[ERTS_SMALL_FIX_MSG_SZ-1]; } ErtsSmallFixSzMessage; typedef struct { ErtsMessage m; Eterm data[ERTS_MEDIUM_FIX_MSG_SZ-1]; } ErtsMediumFixSzMessage; typedef struct { ErtsMessage m; Eterm data[ERTS_LARGE_FIX_MSG_SZ-1]; } ErtsLargeFixSzMessage; ErtsMessage *erts_try_alloc_message_on_heap(Process *pp, erts_aint32_t *psp, ErtsProcLocks *plp, Uint sz, Eterm **hpp, ErlOffHeap **ohpp, int *on_heap_p); ErtsMessage *erts_realloc_shrink_message(ErtsMessage *mp, Uint sz, Eterm *brefs, Uint brefs_size); ERTS_GLB_FORCE_INLINE ErtsMessage *erts_alloc_message(Uint sz, Eterm **hpp); ERTS_GLB_FORCE_INLINE ErtsMessage *erts_shrink_message(ErtsMessage *mp, Uint sz, Eterm *brefs, Uint brefs_size); ERTS_GLB_FORCE_INLINE void erts_free_message(ErtsMessage *mp); ERTS_GLB_INLINE Uint erts_used_frag_sz(const ErlHeapFragment*); ERTS_GLB_INLINE Uint erts_msg_attached_data_size(ErtsMessage *msg); #define ERTS_MSG_COMBINED_HFRAG ((void *) 0x1) #define erts_message_to_heap_frag(MP) \ (((MP)->data.attached == ERTS_MSG_COMBINED_HFRAG) ? \ &(MP)->hfrag : (MP)->data.heap_frag) #if ERTS_GLB_INLINE_INCL_FUNC_DEF ERTS_GLB_FORCE_INLINE ErtsMessage *erts_alloc_message(Uint sz, Eterm **hpp) { ErtsMessage *mp; if (sz == 0) { mp = erts_alloc_message_ref(); ERTS_INIT_MESSAGE(mp); if (hpp) *hpp = NULL; return mp; } mp = erts_alloc(ERTS_ALC_T_MSG, sizeof(ErtsMessage) + (sz - 1)*sizeof(Eterm)); ERTS_INIT_MESSAGE(mp); mp->data.attached = ERTS_MSG_COMBINED_HFRAG; ERTS_INIT_HEAP_FRAG(&mp->hfrag, sz, sz); if (hpp) *hpp = &mp->hfrag.mem[0]; return mp; } ERTS_GLB_FORCE_INLINE ErtsMessage * erts_shrink_message(ErtsMessage *mp, Uint sz, Eterm *brefs, Uint brefs_size) { if (sz == 0) { ErtsMessage *nmp; if (!mp->data.attached) return mp; ASSERT(mp->data.attached == ERTS_MSG_COMBINED_HFRAG); nmp = erts_alloc_message_ref(); #ifdef DEBUG if (brefs && brefs_size) { int i; for (i = 0; i < brefs_size; i++) ASSERT(is_non_value(brefs[i]) || is_immed(brefs[i])); } #endif erts_free(ERTS_ALC_T_MSG, mp); return nmp; } ASSERT(mp->data.attached == ERTS_MSG_COMBINED_HFRAG); ASSERT(mp->hfrag.used_size >= sz); if (sz >= (mp->hfrag.alloc_size - mp->hfrag.alloc_size / 16)) { mp->hfrag.used_size = sz; return mp; } return erts_realloc_shrink_message(mp, sz, brefs, brefs_size); } ERTS_GLB_FORCE_INLINE void erts_free_message(ErtsMessage *mp) { if (mp->data.attached != ERTS_MSG_COMBINED_HFRAG) erts_free_message_ref(mp); else erts_free(ERTS_ALC_T_MSG, mp); } ERTS_GLB_INLINE Uint erts_used_frag_sz(const ErlHeapFragment* bp) { Uint sz = 0; for ( ; bp!=NULL; bp=bp->next) { sz += bp->used_size; } return sz; } ERTS_GLB_INLINE Uint erts_msg_attached_data_size(ErtsMessage *msg) { ASSERT(msg->data.attached); if (ERTS_SIG_IS_INTERNAL_MSG(msg)) return erts_used_frag_sz(erts_message_to_heap_frag(msg)); return erts_msg_attached_data_size_aux(msg); } #endif Uint erts_mbuf_size(Process *p); #if defined(DEBUG) || 0 # define ERTS_CHK_MBUF_SZ(P) \ do { \ Uint actual_mbuf_sz__ = erts_mbuf_size((P)); \ ERTS_ASSERT((P)->mbuf_sz >= actual_mbuf_sz__); \ } while (0) #else # define ERTS_CHK_MBUF_SZ(P) ((void) 1) #endif #define ERTS_FOREACH_SIG_PRIVQS(PROC, MVAR, CODE) \ do { \ int i__; \ ErtsMessage *msgs__[] = {(PROC)->sig_qs.first, \ (PROC)->sig_qs.cont}; \ for (i__ = 0; i__ < sizeof(msgs__)/sizeof(msgs__[0]); i__++) { \ ErtsMessage *MVAR; \ for (MVAR = msgs__[i__]; MVAR; MVAR = MVAR->next) { \ CODE; \ } \ } \ } while (0) #endif