/* * %CopyrightBegin% * * Copyright Ericsson AB 1996-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 __DIST_H__ #define __DIST_H__ #include "erl_process.h" #include "erl_node_tables.h" #include "zlib.h" #define DFLAG_PUBLISHED 0x01 #define DFLAG_ATOM_CACHE 0x02 #define DFLAG_EXTENDED_REFERENCES 0x04 #define DFLAG_DIST_MONITOR 0x08 #define DFLAG_FUN_TAGS 0x10 #define DFLAG_DIST_MONITOR_NAME 0x20 #define DFLAG_HIDDEN_ATOM_CACHE 0x40 #define DFLAG_NEW_FUN_TAGS 0x80 #define DFLAG_EXTENDED_PIDS_PORTS 0x100 #define DFLAG_EXPORT_PTR_TAG 0x200 #define DFLAG_BIT_BINARIES 0x400 #define DFLAG_NEW_FLOATS 0x800 #define DFLAG_UNICODE_IO 0x1000 #define DFLAG_DIST_HDR_ATOM_CACHE 0x2000 #define DFLAG_SMALL_ATOM_TAGS 0x4000 #define DFLAG_INTERNAL_TAGS 0x8000 /* used by ETS 'compressed' option */ #define DFLAG_UTF8_ATOMS 0x10000 #define DFLAG_MAP_TAG 0x20000 #define DFLAG_BIG_CREATION 0x40000 #define DFLAG_SEND_SENDER 0x80000 #define DFLAG_BIG_SEQTRACE_LABELS 0x100000 #define DFLAG_NO_MAGIC 0x200000 /* internal for pending connection */ #define DFLAG_EXIT_PAYLOAD 0x400000 #define DFLAG_FRAGMENTS 0x800000 /* Mandatory flags for distribution */ #define DFLAG_DIST_MANDATORY (DFLAG_EXTENDED_REFERENCES \ | DFLAG_EXTENDED_PIDS_PORTS \ | DFLAG_UTF8_ATOMS \ | DFLAG_NEW_FUN_TAGS) /* * Additional optimistic flags when encoding toward pending connection. * If remote node (erl_interface) does not supporting these then we may need * to transcode messages enqueued before connection setup was finished. */ #define DFLAG_DIST_HOPEFULLY (DFLAG_EXPORT_PTR_TAG \ | DFLAG_BIT_BINARIES \ | DFLAG_DIST_MONITOR \ | DFLAG_DIST_MONITOR_NAME) /* Our preferred set of flags. Used for connection setup handshake */ #define DFLAG_DIST_DEFAULT (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY \ | DFLAG_FUN_TAGS \ | DFLAG_NEW_FLOATS \ | DFLAG_UNICODE_IO \ | DFLAG_DIST_HDR_ATOM_CACHE \ | DFLAG_SMALL_ATOM_TAGS \ | DFLAG_UTF8_ATOMS \ | DFLAG_MAP_TAG \ | DFLAG_BIG_CREATION \ | DFLAG_SEND_SENDER \ | DFLAG_BIG_SEQTRACE_LABELS \ | DFLAG_EXIT_PAYLOAD \ | DFLAG_FRAGMENTS) /* Flags addable by local distr implementations */ #define DFLAG_DIST_ADDABLE DFLAG_DIST_DEFAULT /* Flags rejectable by local distr implementation */ #define DFLAG_DIST_REJECTABLE (DFLAG_DIST_HDR_ATOM_CACHE \ | DFLAG_HIDDEN_ATOM_CACHE \ | DFLAG_ATOM_CACHE) /* Flags for all features needing strict order delivery */ #define DFLAG_DIST_STRICT_ORDER DFLAG_DIST_HDR_ATOM_CACHE /* All flags that should be enabled when term_to_binary/1 is used. */ #define TERM_TO_BINARY_DFLAGS (DFLAG_EXTENDED_REFERENCES \ | DFLAG_NEW_FUN_TAGS \ | DFLAG_NEW_FLOATS \ | DFLAG_EXTENDED_PIDS_PORTS \ | DFLAG_EXPORT_PTR_TAG \ | DFLAG_BIT_BINARIES \ | DFLAG_MAP_TAG \ | DFLAG_BIG_CREATION) /* opcodes used in distribution messages */ enum dop { DOP_LINK = 1, DOP_SEND = 2, DOP_EXIT = 3, DOP_UNLINK = 4, /* Ancient DOP_NODE_LINK (5) was here, can be reused */ DOP_REG_SEND = 6, DOP_GROUP_LEADER = 7, DOP_EXIT2 = 8, DOP_SEND_TT = 12, DOP_EXIT_TT = 13, DOP_REG_SEND_TT = 16, DOP_EXIT2_TT = 18, DOP_MONITOR_P = 19, DOP_DEMONITOR_P = 20, DOP_MONITOR_P_EXIT = 21, DOP_SEND_SENDER = 22, DOP_SEND_SENDER_TT = 23, /* These are used when DFLAG_EXIT_PAYLOAD is detected */ DOP_PAYLOAD_EXIT = 24, DOP_PAYLOAD_EXIT_TT = 25, DOP_PAYLOAD_EXIT2 = 26, DOP_PAYLOAD_EXIT2_TT = 27, DOP_PAYLOAD_MONITOR_P_EXIT = 28 }; /* distribution trap functions */ extern Export* dmonitor_node_trap; typedef enum { ERTS_DSP_NO_LOCK, ERTS_DSP_RLOCK } ErtsDSigPrepLock; /* Must be larger or equal to 16 */ #ifdef DEBUG #define ERTS_DIST_FRAGMENT_SIZE 16 #else /* This should be made configurable */ #define ERTS_DIST_FRAGMENT_SIZE (64 * 1024) #endif #define ERTS_DIST_FRAGMENT_HEADER_SIZE (1 + 1 + 8 + 8) /* magic, header, seq id, frag id*/ #define ERTS_DE_BUSY_LIMIT (1024*1024) extern int erts_dist_buf_busy_limit; extern int erts_is_alive; /* * erts_dsig_prepare() prepares a send of a distributed signal. * One of the values defined below are returned. */ /* Connected; signals can be enqueued and sent. */ #define ERTS_DSIG_PREP_CONNECTED 0 /* Not connected; connection needs to be set up. */ #define ERTS_DSIG_PREP_NOT_CONNECTED 1 /* Caller would be suspended on send operation. */ #define ERTS_DSIG_PREP_WOULD_SUSPEND 2 /* System not alive (distributed) */ #define ERTS_DSIG_PREP_NOT_ALIVE 3 /* Pending connection; signals can be enqueued */ #define ERTS_DSIG_PREP_PENDING 4 /* dist_ctrl_{g,s}et_option/2 */ #define ERTS_DIST_CTRL_OPT_GET_SIZE ((Uint32) (1 << 0)) #ifdef DEBUG #define ERTS_DBG_CHK_NO_DIST_LNK(D, R, L) \ erts_dbg_chk_no_dist_proc_link((D), (R), (L)) #else #define ERTS_DBG_CHK_NO_DIST_LNK(D, R, L) #endif /* Define for testing */ /* #define EXTREME_TTB_TRAPPING 1 */ #ifndef EXTREME_TTB_TRAPPING #define TERM_TO_BINARY_LOOP_FACTOR 32 #else #define TERM_TO_BINARY_LOOP_FACTOR 1 #endif typedef enum { TTBSize, TTBEncode, TTBCompress } TTBState; typedef struct TTBSizeContext_ { Uint flags; int level; Uint result; Eterm obj; ErtsWStack wstack; } TTBSizeContext; typedef struct TTBEncodeContext_ { Uint flags; Uint hopefull_flags; int level; byte* ep; Eterm obj; ErtsWStack wstack; Binary *result_bin; } TTBEncodeContext; typedef struct { Uint real_size; Uint dest_len; byte *dbytes; Binary *result_bin; Binary *destination_bin; z_stream stream; } TTBCompressContext; typedef struct { int alive; TTBState state; union { TTBSizeContext sc; TTBEncodeContext ec; TTBCompressContext cc; } s; } TTBContext; enum erts_dsig_send_phase { ERTS_DSIG_SEND_PHASE_INIT, ERTS_DSIG_SEND_PHASE_MSG_SIZE, ERTS_DSIG_SEND_PHASE_ALLOC, ERTS_DSIG_SEND_PHASE_MSG_ENCODE, ERTS_DSIG_SEND_PHASE_FIN, ERTS_DSIG_SEND_PHASE_SEND }; typedef struct erts_dsig_send_context { int connect; int no_suspend; int no_trap; Eterm ctl; Eterm msg; Eterm from; Eterm ctl_heap[6]; Eterm return_term; DistEntry *dep; Eterm node; /* used if dep == NULL */ Eterm cid; Eterm connection_id; int deref_dep; enum erts_dsig_send_phase phase; Sint reds; Uint32 max_finalize_prepend; Uint data_size, dhdr_ext_size; ErtsAtomCacheMap *acmp; ErtsDistOutputBuf *obuf; Uint fragments; Uint32 flags; Process *c_p; union { TTBSizeContext sc; TTBEncodeContext ec; }u; } ErtsDSigSendContext; typedef struct dist_sequences DistSeqNode; struct dist_sequences { ErlHeapFragment hfrag; struct dist_sequences *parent; struct dist_sequences *left; struct dist_sequences *right; char is_red; Uint64 seq_id; int cnt; Sint ctl_len; }; /* * erts_dsig_send_* return values. */ #define ERTS_DSIG_SEND_OK 0 #define ERTS_DSIG_SEND_YIELD 1 #define ERTS_DSIG_SEND_CONTINUE 2 #define ERTS_DSIG_SEND_TOO_LRG 3 extern int erts_dsig_send_msg(ErtsDSigSendContext*, Eterm, Eterm); extern int erts_dsig_send_reg_msg(ErtsDSigSendContext*, Eterm, Eterm); extern int erts_dsig_send_link(ErtsDSigSendContext *, Eterm, Eterm); extern int erts_dsig_send_exit_tt(ErtsDSigSendContext *, Eterm, Eterm, Eterm, Eterm); extern int erts_dsig_send_unlink(ErtsDSigSendContext *, Eterm, Eterm); extern int erts_dsig_send_group_leader(ErtsDSigSendContext *, Eterm, Eterm); extern int erts_dsig_send_exit(ErtsDSigSendContext *, Eterm, Eterm, Eterm); extern int erts_dsig_send_exit2(ErtsDSigSendContext *, Eterm, Eterm, Eterm); extern int erts_dsig_send_demonitor(ErtsDSigSendContext *, Eterm, Eterm, Eterm); extern int erts_dsig_send_monitor(ErtsDSigSendContext *, Eterm, Eterm, Eterm); extern int erts_dsig_send_m_exit(ErtsDSigSendContext *, Eterm, Eterm, Eterm, Eterm); extern int erts_dsig_send(ErtsDSigSendContext *dsdp); extern int erts_dsend_context_dtor(Binary*); extern Eterm erts_dsend_export_trap_context(Process* p, ErtsDSigSendContext* ctx); extern int erts_dist_command(Port *prt, int reds); extern void erts_dist_port_not_busy(Port *prt); extern void erts_kill_dist_connection(DistEntry *dep, Uint32); extern Uint erts_dist_cache_size(void); extern Sint erts_abort_connection_rwunlock(DistEntry *dep); extern void erts_debug_dist_seq_tree_foreach( DistEntry *dep, int (*func)(DistSeqNode *, void*, Sint), void *args); extern int erts_dsig_prepare(ErtsDSigSendContext *, DistEntry*, Process *, ErtsProcLocks, ErtsDSigPrepLock, int, int, int); void erts_dist_print_procs_suspended_on_de(fmtfn_t to, void *to_arg); int erts_auto_connect(DistEntry* dep, Process *proc, ErtsProcLocks proc_locks); #endif