path: root/erts/emulator/beam/dist.h

/*
 * %CopyrightBegin%
 *
 * Copyright Ericsson AB 1996-2017. 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
#define DFLAG_UTF8_ATOMS          0x10000
#define DFLAG_MAP_TAG             0x20000
#define DFLAG_BIG_CREATION        0x40000
#define DFLAG_SEND_SENDER         0x80000
#define DFLAG_NO_MAGIC            0x100000

/* Mandatory flags for distribution (sync with dist_util.erl) */
#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 */
#define DFLAG_DIST_HOPEFULLY (DFLAG_NO_MAGIC                    \
                              | DFLAG_EXPORT_PTR_TAG            \
                              | DFLAG_BIT_BINARIES)

/* 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 */
#define DOP_LINK		1
#define DOP_SEND		2
#define DOP_EXIT		3
#define DOP_UNLINK		4
/* Ancient DOP_NODE_LINK (5) was here, can be reused */
#define DOP_REG_SEND		6
#define DOP_GROUP_LEADER	7
#define DOP_EXIT2		8

#define DOP_SEND_TT		12
#define DOP_EXIT_TT		13
#define DOP_REG_SEND_TT		16
#define DOP_EXIT2_TT		18

#define DOP_MONITOR_P		19
#define DOP_DEMONITOR_P		20
#define DOP_MONITOR_P_EXIT	21

#define DOP_SEND_SENDER         22
#define DOP_SEND_SENDER_TT      23

/* distribution trap functions */
extern Export* dmonitor_node_trap;
extern Export* dmonitor_p_trap;

typedef enum {
    ERTS_DSP_NO_LOCK,
    ERTS_DSP_RLOCK,
    ERTS_DSP_RWLOCK
} ErtsDSigPrepLock;


typedef struct {
    Process *proc;
    DistEntry *dep;
    Eterm node;   /* used if dep == NULL */
    Eterm cid;
    Eterm connection_id;
    int no_suspend;
} ErtsDSigData;

#define ERTS_DE_IS_NOT_CONNECTED(DEP) \
  (ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&(DEP)->rwmtx) \
		      || erts_lc_rwmtx_is_rwlocked(&(DEP)->rwmtx)), \
   (is_nil((DEP)->cid) || ((DEP)->status & ERTS_DE_SFLG_EXITING)))

#define ERTS_DE_IS_CONNECTED(DEP) \
  (!ERTS_DE_IS_NOT_CONNECTED((DEP)))

#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

ERTS_GLB_INLINE int erts_dsig_prepare(ErtsDSigData *,
				      DistEntry **,
				      Process *,
                                      ErtsProcLocks,
				      ErtsDSigPrepLock,
				      int,
				      int);

ERTS_GLB_INLINE
void erts_schedule_dist_command(Port *, DistEntry *);

#if ERTS_GLB_INLINE_INCL_FUNC_DEF

ERTS_GLB_INLINE int 
erts_dsig_prepare(ErtsDSigData *dsdp,
		  DistEntry **depp,
		  Process *proc,
                  ErtsProcLocks proc_locks,
		  ErtsDSigPrepLock dspl,
		  int no_suspend,
		  int connect)
{
    DistEntry* dep = *depp;
    int deref_dep = 0;
    int res;

    if (!erts_is_alive)
	return ERTS_DSIG_PREP_NOT_ALIVE;
    if (!dep) {
	if (!connect)
	    return ERTS_DSIG_PREP_NOT_CONNECTED;

	dep = erts_find_or_insert_dist_entry(dsdp->node);
	ASSERT(dep != erts_this_dist_entry);  /* SVERK: What to do? */
        deref_dep = 1;
    }

#ifdef ERTS_ENABLE_LOCK_CHECK
    if (connect) {
        erts_proc_lc_might_unlock(proc, proc_locks);
    }
#endif

retry:
    if (dspl == ERTS_DSP_RWLOCK)
	erts_de_rwlock(dep);
    else
	erts_de_rlock(dep);

    if (ERTS_DE_IS_CONNECTED(dep)) {
	res = ERTS_DSIG_PREP_CONNECTED;
    }
    else if (dep->status & ERTS_DE_SFLG_PENDING) {
	res = ERTS_DSIG_PREP_PENDING;
    }
    else if (dep->status & ERTS_DE_SFLG_EXITING) {
	/* SVERK is this ok, or should we trigger another connection setup */
	res = ERTS_DSIG_PREP_NOT_CONNECTED;
	goto fail;
    }
    else if (connect) {
        ASSERT(dep->status == 0);
	if (dspl != ERTS_DSP_RWLOCK) {
	    erts_de_runlock(dep);
	    erts_de_rwlock(dep);
	}
	if (dep->status == 0) {
	    Process* net_kernel;
	    ErtsProcLocks nk_locks = ERTS_PROC_LOCK_MSGQ;
	    Eterm *hp;
	    ErlOffHeap *ohp;
	    ErtsMessage *mp;
	    Eterm msg, conn_id, dhandle;

	    dep->status = ERTS_DE_SFLG_PENDING;
	    dep->flags = (DFLAG_DIST_MANDATORY | DFLAG_DIST_HOPEFULLY);
	    dep->connection_id++;
	    dep->connection_id &= ERTS_DIST_CON_ID_MASK;
            conn_id = make_small(dep->connection_id);
	    erts_de_rwunlock(dep);

	    net_kernel = erts_whereis_process(proc, proc_locks,
					      am_net_kernel, nk_locks, 0);
	    if (!net_kernel) {
                if (deref_dep)
                    erts_deref_dist_entry(dep);
		return ERTS_DSIG_PREP_NOT_ALIVE;
	    }

	    /* Send {auto_connect, Node, ConnId, DHandle} to net_kernel */
	    mp = erts_alloc_message_heap(net_kernel, &nk_locks,
                                         5 + ERTS_MAGIC_REF_THING_SIZE,
                                         &hp, &ohp);
            dhandle = erts_build_dhandle(&hp, ohp, dep);
	    msg = TUPLE4(hp, am_auto_connect, dep->sysname, conn_id, dhandle);
            erts_queue_message(net_kernel, nk_locks, mp, msg, proc->common.id);
	    erts_proc_unlock(net_kernel, nk_locks);
	}
	else
	    erts_de_rwunlock(dep);
	goto retry;
    }
    else {
        ASSERT(dep->status == 0);
	res = ERTS_DSIG_PREP_NOT_CONNECTED;
	goto fail;
    }

    if (no_suspend) {
	if (erts_atomic32_read_acqb(&dep->qflgs) & ERTS_DE_QFLG_BUSY) {
	    res = ERTS_DSIG_PREP_WOULD_SUSPEND;
	    goto fail;
	}
    }
    dsdp->proc = proc;
    dsdp->dep = dep;
    dsdp->cid = dep->cid;
    dsdp->connection_id = dep->connection_id;
    dsdp->no_suspend = no_suspend;
    if (dspl == ERTS_DSP_NO_LOCK)
	erts_de_runlock(dep);
    if (deref_dep)
        erts_deref_dist_entry(dep);
    *depp = dep;
    return res;

 fail:
    if (dspl == ERTS_DSP_RWLOCK)
	erts_de_rwunlock(dep);
    else
	erts_de_runlock(dep);
    if (deref_dep)
        erts_deref_dist_entry(dep);
    return res;
}

ERTS_GLB_INLINE
void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry)
{
    DistEntry *dep;
    Eterm id;

    if (prt) {
	ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
	ASSERT((erts_atomic32_read_nob(&prt->state)
		& ERTS_PORT_SFLGS_DEAD) == 0);
	ASSERT(prt->dist_entry);

	dep = prt->dist_entry;
	id = prt->common.id;
    }
    else {
	ASSERT(dist_entry);
	ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&dist_entry->rwmtx)
			   || erts_lc_rwmtx_is_rwlocked(&dist_entry->rwmtx));
	ASSERT(is_internal_port(dist_entry->cid));

 	dep = dist_entry;
	id = dep->cid;
    }

    if (!erts_atomic_xchg_mb(&dep->dist_cmd_scheduled, 1))
	erts_port_task_schedule(id, &dep->dist_cmd, ERTS_PORT_TASK_DIST_CMD);
}

#endif

typedef struct {
    ErtsLink *d_lnk;
    ErtsLink *d_sub_lnk;
} ErtsDistLinkData;

ERTS_GLB_INLINE void erts_remove_dist_link(ErtsDistLinkData *,
					   Eterm,
					   Eterm,
					   DistEntry *);
ERTS_GLB_INLINE int erts_was_dist_link_removed(ErtsDistLinkData *);
ERTS_GLB_INLINE void erts_destroy_dist_link(ErtsDistLinkData *);

#if ERTS_GLB_INLINE_INCL_FUNC_DEF

ERTS_GLB_INLINE void
erts_remove_dist_link(ErtsDistLinkData *dldp,
		      Eterm lid,
		      Eterm rid,
		      DistEntry *dep)
{
    erts_de_links_lock(dep);
    dldp->d_lnk = erts_lookup_link(dep->nlinks, lid);
    if (!dldp->d_lnk)
	dldp->d_sub_lnk = NULL;
    else {
	dldp->d_sub_lnk = erts_remove_link(&ERTS_LINK_ROOT(dldp->d_lnk), rid);
	dldp->d_lnk = (ERTS_LINK_ROOT(dldp->d_lnk)
		       ? NULL
		       : erts_remove_link(&dep->nlinks, lid));
    }
    erts_de_links_unlock(dep);
}

ERTS_GLB_INLINE int
erts_was_dist_link_removed(ErtsDistLinkData *dldp)
{
    return dldp->d_sub_lnk != NULL;
}

ERTS_GLB_INLINE void
erts_destroy_dist_link(ErtsDistLinkData *dldp)
{
    if (dldp->d_lnk)
	erts_destroy_link(dldp->d_lnk);
    if (dldp->d_sub_lnk)
	erts_destroy_link(dldp->d_sub_lnk);	
}

#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;
    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
};

struct erts_dsig_send_context {
    enum erts_dsig_send_phase phase;
    Sint reds;

    Eterm ctl;
    Eterm msg;
    int force_busy;
    Uint32 max_finalize_prepend;
    Uint data_size, dhdr_ext_size;
    ErtsAtomCacheMap *acmp;
    ErtsDistOutputBuf *obuf;
    Uint32 flags;
    Process *c_p;
    union {
	TTBSizeContext sc;
	TTBEncodeContext ec;
    }u;
};

typedef struct {
    int suspend;
    int connect;

    Eterm ctl_heap[6];
    ErtsDSigData dsd;
    DistEntry *dep;
    int deref_dep;
    struct erts_dsig_send_context dss;

    Eterm return_term;
}ErtsSendContext;


/*
 * erts_dsig_send_* return values.
 */
#define ERTS_DSIG_SEND_OK	0
#define ERTS_DSIG_SEND_YIELD	1
#define ERTS_DSIG_SEND_CONTINUE 2

extern int erts_dsig_send_link(ErtsDSigData *, Eterm, Eterm);
extern int erts_dsig_send_msg(Eterm, Eterm, ErtsSendContext*);
extern int erts_dsig_send_exit_tt(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
extern int erts_dsig_send_unlink(ErtsDSigData *, Eterm, Eterm);
extern int erts_dsig_send_reg_msg(Eterm, Eterm, ErtsSendContext*);
extern int erts_dsig_send_group_leader(ErtsDSigData *, Eterm, Eterm);
extern int erts_dsig_send_exit(ErtsDSigData *, Eterm, Eterm, Eterm);
extern int erts_dsig_send_exit2(ErtsDSigData *, Eterm, Eterm, Eterm);
extern int erts_dsig_send_demonitor(ErtsDSigData *, Eterm, Eterm, Eterm, int);
extern int erts_dsig_send_monitor(ErtsDSigData *, Eterm, Eterm, Eterm);
extern int erts_dsig_send_m_exit(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);

extern int erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx);
extern int erts_dsend_context_dtor(Binary*);
extern Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* 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);


#endif