diff options
Diffstat (limited to 'erts/emulator/beam/external.c')
-rw-r--r-- | erts/emulator/beam/external.c | 1210 |
1 files changed, 808 insertions, 402 deletions
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c index c0a3838d42..ec67ab2aed 100644 --- a/erts/emulator/beam/external.c +++ b/erts/emulator/beam/external.c @@ -1,7 +1,7 @@ /* * %CopyrightBegin% * - * Copyright Ericsson AB 1996-2017. All Rights Reserved. + * 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. @@ -102,7 +102,7 @@ static byte* enc_term(ErtsAtomCacheMap *, Eterm, byte*, Uint32, struct erl_off_h struct TTBEncodeContext_; static int enc_term_int(struct TTBEncodeContext_*,ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, Uint32 dflags, struct erl_off_heap_header** off_heap, Sint *reds, byte **res); -static Uint is_external_string(Eterm obj, int* p_is_string); +static int is_external_string(Eterm obj, Uint* lenp); static byte* enc_atom(ErtsAtomCacheMap *, Eterm, byte*, Uint32); static byte* enc_pid(ErtsAtomCacheMap *, Eterm, byte*, Uint32); struct B2TContext_t; @@ -112,18 +112,19 @@ static byte* dec_pid(ErtsDistExternal *, ErtsHeapFactory*, byte*, Eterm*, byte t static Sint decoded_size(byte *ep, byte* endp, int internal_tags, struct B2TContext_t*); static BIF_RETTYPE term_to_binary_trap_1(BIF_ALIST_1); -static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint flags, - Binary *context_b); +static Eterm erts_term_to_binary_int(Process* p, Eterm Term, Eterm opts, int level, + Uint flags, Binary *context_b); static Uint encode_size_struct2(ErtsAtomCacheMap *, Eterm, unsigned); struct TTBSizeContext_; -static int encode_size_struct_int(struct TTBSizeContext_*, ErtsAtomCacheMap *acmp, Eterm obj, - unsigned dflags, Sint *reds, Uint *res); +static ErtsExtSzRes encode_size_struct_int(struct TTBSizeContext_*, ErtsAtomCacheMap *acmp, + Eterm obj, unsigned dflags, Sint *reds, Uint *res); static Export binary_to_term_trap_export; static BIF_RETTYPE binary_to_term_trap_1(BIF_ALIST_1); -static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binary* context_b, - Export *bif, Eterm arg0, Eterm arg1); +static Sint transcode_dist_obuf(ErtsDistOutputBuf*, DistEntry*, Uint32 dflags, Sint reds); + + void erts_init_external(void) { erts_init_trap_export(&term_to_binary_trap_export, @@ -222,23 +223,10 @@ erts_destroy_atom_cache_map(ErtsAtomCacheMap *acmp) static ERTS_INLINE void insert_acache_map(ErtsAtomCacheMap *acmp, Eterm atom, Uint32 dflags) { - /* - * If the receiver do not understand utf8 atoms - * and this atom cannot be represented in latin1, - * we are not allowed to cache it. - * - * In this case all atoms are assumed to have - * latin1 encoding in the cache. By refusing it - * in the cache we will instead encode it using - * ATOM_UTF8_EXT/SMALL_ATOM_UTF8_EXT which the - * receiver do not recognize and tear down the - * connection. - */ - if (acmp && acmp->sz < ERTS_MAX_INTERNAL_ATOM_CACHE_ENTRIES - && ((dflags & DFLAG_UTF8_ATOMS) - || atom_tab(atom_val(atom))->latin1_chars >= 0)) { + if (acmp && acmp->sz < ERTS_MAX_INTERNAL_ATOM_CACHE_ENTRIES) { int ix; ASSERT(acmp->hdr_sz < 0); + ASSERT(dflags & DFLAG_UTF8_ATOMS); ix = atom2cix(atom); if (acmp->cache[ix].iix < 0) { acmp->cache[ix].iix = acmp->sz; @@ -258,9 +246,7 @@ get_iix_acache_map(ErtsAtomCacheMap *acmp, Eterm atom, Uint32 dflags) ASSERT(is_atom(atom)); ix = atom2cix(atom); if (acmp->cache[ix].iix < 0) { - ASSERT(acmp->sz == ERTS_MAX_INTERNAL_ATOM_CACHE_ENTRIES - || (!(dflags & DFLAG_UTF8_ATOMS) - && atom_tab(atom_val(atom))->latin1_chars < 0)); + ASSERT(acmp->sz == ERTS_MAX_INTERNAL_ATOM_CACHE_ENTRIES); return -1; } else { @@ -274,21 +260,14 @@ void erts_finalize_atom_cache_map(ErtsAtomCacheMap *acmp, Uint32 dflags) { if (acmp) { - int utf8_atoms = (int) (dflags & DFLAG_UTF8_ATOMS); int long_atoms = 0; /* !0 if one or more atoms are longer than 255. */ int i; - int sz; - int fix_sz - = 1 /* VERSION_MAGIC */ - + 1 /* DIST_HEADER */ - + 1 /* dist header flags */ - + 1 /* number of internal cache entries */ - ; + int sz = 0; int min_sz; + ASSERT(dflags & DFLAG_UTF8_ATOMS); ASSERT(acmp->hdr_sz < 0); /* Make sure cache update instructions fit */ - min_sz = fix_sz+(2+4)*acmp->sz; - sz = fix_sz; + min_sz = (2+4)*acmp->sz; for (i = 0; i < acmp->sz; i++) { Atom *a; Eterm atom; @@ -296,7 +275,7 @@ erts_finalize_atom_cache_map(ErtsAtomCacheMap *acmp, Uint32 dflags) atom = acmp->cache[acmp->cix[i]].atom; ASSERT(is_atom(atom)); a = atom_tab(atom_val(atom)); - len = (int) (utf8_atoms ? a->len : a->latin1_chars); + len = (int) a->len; ASSERT(len >= 0); if (!long_atoms && len > 255) long_atoms = 1; @@ -316,17 +295,28 @@ erts_finalize_atom_cache_map(ErtsAtomCacheMap *acmp, Uint32 dflags) } Uint -erts_encode_ext_dist_header_size(ErtsAtomCacheMap *acmp) +erts_encode_ext_dist_header_size(ErtsAtomCacheMap *acmp, Uint fragments) { if (!acmp) return 0; else { + int fix_sz + = 1 /* VERSION_MAGIC */ + + 1 /* DIST_HEADER */ + + 1 /* dist header flags */ + + 1 /* number of internal cache entries */ + ; ASSERT(acmp->hdr_sz >= 0); - return acmp->hdr_sz; + if (fragments > 1) + fix_sz += 8 /* sequence id */ + + 8 /* number of fragments */ + ; + return fix_sz + acmp->hdr_sz; } } -byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp) +byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp, + Uint fragments, Eterm from) { /* Maximum number of atom must be less than the maximum of a 32 bits unsigned integer. Check is done in erl_init.c, erl_start function. */ @@ -360,24 +350,120 @@ byte *erts_encode_ext_dist_header_setup(byte *ctl_ext, ErtsAtomCacheMap *acmp) put_int8(acmp->sz, ep); --ep; put_int8(dist_hdr_flags, ep); - *--ep = DIST_HEADER; - *--ep = VERSION_MAGIC; + if (fragments > 1) { + ASSERT(is_pid(from)); + ep -= 8; + put_int64(fragments, ep); + ep -= 8; + put_int64(from, ep); + *--ep = DIST_FRAG_HEADER; + } else { + *--ep = DIST_HEADER; + } + *--ep = VERSION_MAGIC; return ep; } } -byte *erts_encode_ext_dist_header_finalize(byte *ext, ErtsAtomCache *cache, Uint32 dflags) +byte *erts_encode_ext_dist_header_fragment(byte **hdrpp, + Uint fragment, + Eterm from) +{ + byte *ep = *hdrpp, *start = ep; + ASSERT(is_pid(from)); + *ep++ = VERSION_MAGIC; + *ep++ = DIST_FRAG_CONT; + put_int64(from, ep); + ep += 8; + put_int64(fragment, ep); + ep += 8; + *hdrpp = ep; + return start; +} + + +#define PASS_THROUGH 'p' + +Sint erts_encode_ext_dist_header_finalize(ErtsDistOutputBuf* ob, + DistEntry* dep, + Uint32 dflags, + Sint reds) { byte *ip; byte instr_buf[(2+4)*ERTS_ATOM_CACHE_SIZE]; int ci, sz; byte dist_hdr_flags; int long_atoms; - int utf8_atoms = (int) (dflags & DFLAG_UTF8_ATOMS); - register byte *ep = ext; - ASSERT(ep[0] == VERSION_MAGIC); - if (ep[1] != DIST_HEADER) - return ext; + Uint64 seq_id = 0, frag_id = 0; + register byte *ep = ob->hdrp ? ob->hdrp : ob->extp; + ASSERT(dflags & DFLAG_UTF8_ATOMS); + + /* + * The buffer can have different layouts at this point depending on + * what was known when encoded: + * + * Pending connection: CtrlTerm [, MsgTerm] + * With atom cache : VERSION_MAGIC, DIST_HEADER, ..., CtrlTerm [, MsgTerm] + * No atom cache : VERSION_MAGIC, CtrlTerm [, VERSION_MAGIC, MsgTerm] + */ + + if (ep[0] != VERSION_MAGIC || dep->transcode_ctx) { + /* + * Was encoded without atom cache toward pending connection. + */ + ASSERT(ep[0] == SMALL_TUPLE_EXT || ep[0] == LARGE_TUPLE_EXT); + + if (~dflags & (DFLAG_DIST_MONITOR | DFLAG_DIST_MONITOR_NAME) + && ep[0] == SMALL_TUPLE_EXT + && ep[1] == 4 + && ep[2] == SMALL_INTEGER_EXT + && (ep[3] == DOP_MONITOR_P || + ep[3] == DOP_MONITOR_P_EXIT || + ep[3] == DOP_DEMONITOR_P)) { + /* + * Receiver does not support process monitoring. + * Suppress monitor control msg (see erts_dsig_send_monitor) + * by converting it to an empty (tick) packet. + */ + ob->ext_endp = ob->extp; + return reds; + } + if (~dflags & (DFLAG_BIT_BINARIES | DFLAG_EXPORT_PTR_TAG + | DFLAG_DIST_HDR_ATOM_CACHE)) { + reds = transcode_dist_obuf(ob, dep, dflags, reds); + if (reds < 0) + return reds; + ep = ob->extp; + } + if (dflags & DFLAG_DIST_HDR_ATOM_CACHE) { + /* + * Encoding was done without atom caching but receiver expects + * a dist header, so we prepend an empty one. + */ + *--ep = 0; /* NumberOfAtomCacheRefs */ + *--ep = DIST_HEADER; + *--ep = VERSION_MAGIC; + } + goto done; + } + else if (ep[1] != DIST_HEADER && ep[1] != DIST_FRAG_HEADER && ep[1] != DIST_FRAG_CONT) { + ASSERT(ep[1] == SMALL_TUPLE_EXT || ep[1] == LARGE_TUPLE_EXT); + ASSERT(!(dflags & DFLAG_DIST_HDR_ATOM_CACHE)); + /* Node without atom cache, 'pass through' needed */ + *--ep = PASS_THROUGH; + goto done; + } + + if (ep[1] == DIST_FRAG_CONT) { + ep = ob->extp; + goto done; + } else if (ep[1] == DIST_FRAG_HEADER) { + /* skip the seq id and frag id */ + seq_id = get_int64(&ep[2]); + ep += 8; + frag_id = get_int64(&ep[2]); + ep += 8; + } dist_hdr_flags = ep[2]; long_atoms = ERTS_DIST_HDR_LONG_ATOMS_FLG & ((int) dist_hdr_flags); @@ -405,6 +491,7 @@ byte *erts_encode_ext_dist_header_finalize(byte *ext, ErtsAtomCache *cache, Uint / sizeof(Uint32))+1]; register Uint32 flgs; int iix, flgs_bytes, flgs_buf_ix, used_half_bytes; + ErtsAtomCache* cache = dep->cache; #ifdef DEBUG int tot_used_half_bytes; #endif @@ -447,17 +534,9 @@ byte *erts_encode_ext_dist_header_finalize(byte *ext, ErtsAtomCache *cache, Uint Atom *a; cache->out_arr[cix] = atom; a = atom_tab(atom_val(atom)); - if (utf8_atoms) { - sz = a->len; - ep -= sz; - sys_memcpy((void *) ep, (void *) a->name, sz); - } - else { - ASSERT(0 <= a->latin1_chars && a->latin1_chars <= MAX_ATOM_CHARACTERS); - ep -= a->latin1_chars; - sz = erts_utf8_to_latin1(ep, a->name, a->len); - ASSERT(a->latin1_chars == sz); - } + sz = a->len; + ep -= sz; + sys_memcpy((void *) ep, (void *) a->name, sz); if (long_atoms) { ep -= 2; put_int16(sz, ep); @@ -504,57 +583,70 @@ byte *erts_encode_ext_dist_header_finalize(byte *ext, ErtsAtomCache *cache, Uint break; } } + reds -= 3; /*was ERTS_PORT_REDS_DIST_CMD_FINALIZE*/ } --ep; put_int8(ci, ep); - *--ep = DIST_HEADER; + if (seq_id) { + ep -= 8; + put_int64(frag_id, ep); + ep -= 8; + put_int64(seq_id, ep); + *--ep = DIST_FRAG_HEADER; + } else { + *--ep = DIST_HEADER; + } *--ep = VERSION_MAGIC; - return ep; +done: + ob->extp = ep; + ASSERT((byte*)ob->bin->orig_bytes <= ob->extp && ob->extp < ob->ext_endp); + return reds < 0 ? 0 : reds; } -int erts_encode_dist_ext_size(Eterm term, Uint32 flags, ErtsAtomCacheMap *acmp, - Uint* szp) +ErtsExtSzRes +erts_encode_dist_ext_size(Eterm term, Uint32 flags, ErtsAtomCacheMap *acmp, Uint* szp) { Uint sz; - if (encode_size_struct_int(NULL, acmp, term, flags, NULL, &sz)) { - return -1; - } else { + ErtsExtSzRes res = encode_size_struct_int(NULL, acmp, term, flags, NULL, &sz); + if (res == ERTS_EXT_SZ_OK) { #ifndef ERTS_DEBUG_USE_DIST_SEP - if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE)) + if (!(flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_NO_MAGIC))) #endif sz++ /* VERSION_MAGIC */; *szp += sz; - return 0; } + return res; } -int erts_encode_dist_ext_size_int(Eterm term, struct erts_dsig_send_context* ctx, Uint* szp) +ErtsExtSzRes +erts_encode_dist_ext_size_ctx(Eterm term, ErtsDSigSendContext *ctx, Uint* szp) { Uint sz; - if (encode_size_struct_int(&ctx->u.sc, ctx->acmp, term, ctx->flags, &ctx->reds, &sz)) { - return -1; - } else { + ErtsExtSzRes res = encode_size_struct_int(&ctx->u.sc, ctx->acmp, term, + ctx->flags, &ctx->reds, &sz); + if (res == ERTS_EXT_SZ_OK) { #ifndef ERTS_DEBUG_USE_DIST_SEP - if (!(ctx->flags & DFLAG_DIST_HDR_ATOM_CACHE)) + if (!(ctx->flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_NO_MAGIC))) #endif sz++ /* VERSION_MAGIC */; *szp += sz; - return 0; } + return res; } -Uint erts_encode_ext_size(Eterm term) +ErtsExtSzRes erts_encode_ext_size_2(Eterm term, unsigned dflags, Uint *szp) { - return encode_size_struct2(NULL, term, TERM_TO_BINARY_DFLAGS) - + 1 /* VERSION_MAGIC */; + ErtsExtSzRes res = encode_size_struct_int(NULL, NULL, term, dflags, + NULL, szp); + (*szp)++ /* VERSION_MAGIC */; + return res; } -Uint erts_encode_ext_size_2(Eterm term, unsigned dflags) +ErtsExtSzRes erts_encode_ext_size(Eterm term, Uint *szp) { - return encode_size_struct2(NULL, term, dflags) - + 1 /* VERSION_MAGIC */; + return erts_encode_ext_size_2(term, TERM_TO_BINARY_DFLAGS, szp); } Uint erts_encode_ext_size_ets(Eterm term) @@ -568,7 +660,7 @@ int erts_encode_dist_ext(Eterm term, byte **ext, Uint32 flags, ErtsAtomCacheMap { if (!ctx || !ctx->wstack.wstart) { #ifndef ERTS_DEBUG_USE_DIST_SEP - if (!(flags & DFLAG_DIST_HDR_ATOM_CACHE)) + if (!(flags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_NO_MAGIC))) #endif *(*ext)++ = VERSION_MAGIC; } @@ -593,112 +685,190 @@ byte* erts_encode_ext_ets(Eterm term, byte *ep, struct erl_off_heap_header** off off_heap); } -ErtsDistExternal * -erts_make_dist_ext_copy(ErtsDistExternal *edep, Uint xsize) + +static Uint +dist_ext_size(ErtsDistExternal *edep) { - size_t align_sz; - size_t dist_ext_sz; - size_t ext_sz; - byte *ep; - ErtsDistExternal *new_edep; + Uint sz = sizeof(ErtsDistExternal); - dist_ext_sz = ERTS_DIST_EXT_SIZE(edep); - ASSERT(edep->ext_endp && edep->extp); - ASSERT(edep->ext_endp >= edep->extp); - ext_sz = edep->ext_endp - edep->extp; + ASSERT(edep->data->ext_endp && edep->data->extp); + ASSERT(edep->data->ext_endp >= edep->data->extp); - align_sz = ERTS_EXTRA_DATA_ALIGN_SZ(dist_ext_sz + ext_sz); + if (edep->flags & ERTS_DIST_EXT_ATOM_TRANS_TAB) { + ASSERT(0 <= edep->attab.size \ + && edep->attab.size <= ERTS_ATOM_CACHE_SIZE); + sz -= sizeof(Eterm)*(ERTS_ATOM_CACHE_SIZE - edep->attab.size); + } else { + sz -= sizeof(ErtsAtomTranslationTable); + } + ASSERT(sz % 4 == 0); + return sz; +} - new_edep = erts_alloc(ERTS_ALC_T_EXT_TERM_DATA, - dist_ext_sz + ext_sz + align_sz + xsize); +Uint +erts_dist_ext_size(ErtsDistExternal *edep) +{ + Uint sz = dist_ext_size(edep); + sz += 4; /* may need to pad to 8-byte-align ErtsDistExternalData */ + sz += edep->data[0].frag_id * sizeof(ErtsDistExternalData); + return sz; +} + +Uint +erts_dist_ext_data_size(ErtsDistExternal *edep) +{ + Uint sz = 0, i; + for (i = 0; i < edep->data->frag_id; i++) + sz += edep->data[i].ext_endp - edep->data[i].extp; + return sz; +} + +void +erts_dist_ext_frag(ErtsDistExternalData *ede_datap, ErtsDistExternal *edep) +{ + ErtsDistExternalData *new_ede_datap = &edep->data[edep->data->frag_id - ede_datap->frag_id]; + sys_memcpy(new_ede_datap, ede_datap, sizeof(ErtsDistExternalData)); + + /* If the data is not backed by a binary, we create one here to keep + things simple. Only custom distribution drivers should use lists. */ + if (new_ede_datap->binp == NULL) { + size_t ext_sz = ede_datap->ext_endp - ede_datap->extp; + new_ede_datap->binp = erts_bin_nrml_alloc(ext_sz); + sys_memcpy(new_ede_datap->binp->orig_bytes, (void *) ede_datap->extp, ext_sz); + new_ede_datap->extp = (byte*)new_ede_datap->binp->orig_bytes; + new_ede_datap->ext_endp = (byte*)new_ede_datap->binp->orig_bytes + ext_sz; + } else { + erts_refc_inc(&new_ede_datap->binp->intern.refc, 2); + } +} + +void +erts_make_dist_ext_copy(ErtsDistExternal *edep, ErtsDistExternal *new_edep) +{ + size_t dist_ext_sz = dist_ext_size(edep); + byte *ep; ep = (byte *) new_edep; sys_memcpy((void *) ep, (void *) edep, dist_ext_sz); + erts_ref_dist_entry(new_edep->dep); + ep += dist_ext_sz; - if (new_edep->dep) - erts_smp_refc_inc(&new_edep->dep->refc, 1); - new_edep->extp = ep; - new_edep->ext_endp = ep + ext_sz; - new_edep->heap_size = -1; - sys_memcpy((void *) ep, (void *) edep->extp, ext_sz); - return new_edep; + ep += (UWord)ep & 4; /* 8-byte alignment for ErtsDistExternalData */ + ASSERT((UWord)ep % 8 == 0); + + new_edep->data = (ErtsDistExternalData*)ep; + sys_memzero(new_edep->data, sizeof(ErtsDistExternalData) * edep->data->frag_id); + new_edep->data->frag_id = edep->data->frag_id; + erts_dist_ext_frag(edep->data, new_edep); } -int +void +erts_free_dist_ext_copy(ErtsDistExternal *edep) +{ + int i; + erts_deref_dist_entry(edep->dep); + for (i = 0; i < edep->data->frag_id; i++) + if (edep->data[i].binp) + erts_bin_release(edep->data[i].binp); +} + +ErtsPrepDistExtRes erts_prepare_dist_ext(ErtsDistExternal *edep, byte *ext, Uint size, + Binary *binp, DistEntry *dep, + Uint32 conn_id, ErtsAtomCache *cache) { -#undef ERTS_EXT_FAIL -#undef ERTS_EXT_HDR_FAIL -#if 1 -#define ERTS_EXT_FAIL goto fail -#define ERTS_EXT_HDR_FAIL goto bad_hdr -#else -#define ERTS_EXT_FAIL abort() -#define ERTS_EXT_HDR_FAIL abort() -#endif + register byte *ep; - register byte *ep = ext; - int utf8_atoms = (int) (dep->flags & DFLAG_UTF8_ATOMS); + ASSERT(dep); + erts_de_rlock(dep); + + ASSERT(dep->flags & DFLAG_UTF8_ATOMS); - edep->heap_size = -1; - edep->ext_endp = ext+size; + + if ((dep->state != ERTS_DE_STATE_CONNECTED && + dep->state != ERTS_DE_STATE_PENDING) + || dep->connection_id != conn_id) { + erts_de_runlock(dep); + return ERTS_PREP_DIST_EXT_CLOSED; + } + + if (!(dep->flags & DFLAG_DIST_HDR_ATOM_CACHE)) { + /* Skip PASS_THROUGH */ + ext++; + size--; + } + + ep = ext; if (size < 2) - ERTS_EXT_FAIL; + goto fail; if (ep[0] != VERSION_MAGIC) { erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf(); - if (dep) - erts_dsprintf(dsbufp, - "** Got message from incompatible erlang on " - "channel %d\n", - dist_entry_channel_no(dep)); - else - erts_dsprintf(dsbufp, - "** Attempt to convert old incompatible " - "binary %d\n", - *ep); + erts_dsprintf(dsbufp, + "** Got message from incompatible erlang on " + "channel %d\n", + dist_entry_channel_no(dep)); erts_send_error_to_logger_nogl(dsbufp); - ERTS_EXT_FAIL; + goto fail; } + edep->heap_size = -1; edep->flags = 0; edep->dep = dep; - if (dep) { - erts_smp_de_rlock(dep); - if (dep->flags & DFLAG_DIST_HDR_ATOM_CACHE) - edep->flags |= ERTS_DIST_EXT_DFLAG_HDR; - - edep->flags |= (dep->connection_id & ERTS_DIST_EXT_CON_ID_MASK); - erts_smp_de_runlock(dep); - } + edep->connection_id = conn_id; + edep->data->ext_endp = ext+size; + edep->data->binp = binp; + edep->data->seq_id = 0; + edep->data->frag_id = 1; + + if (dep->flags & DFLAG_DIST_HDR_ATOM_CACHE) + edep->flags |= ERTS_DIST_EXT_DFLAG_HDR; - if (ep[1] != DIST_HEADER) { + if (ep[1] != DIST_HEADER && ep[1] != DIST_FRAG_HEADER && ep[1] != DIST_FRAG_CONT) { if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; edep->attab.size = 0; - edep->extp = ext; + edep->data->extp = ext; + } + else if (ep[1] == DIST_FRAG_CONT) { + if (!(dep->flags & DFLAG_FRAGMENTS)) + goto bad_hdr; + edep->attab.size = 0; + edep->data->extp = ext + 1 + 1 + 8 + 8; + edep->data->seq_id = get_int64(&ep[2]); + edep->data->frag_id = get_int64(&ep[2+8]); + erts_de_runlock(dep); + return ERTS_PREP_DIST_EXT_FRAG_CONT; } else { int tix; int no_atoms; if (!(edep->flags & ERTS_DIST_EXT_DFLAG_HDR)) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; + + if (ep[1] == DIST_FRAG_HEADER) { + if (!(dep->flags & DFLAG_FRAGMENTS)) + goto bad_hdr; + edep->data->seq_id = get_int64(&ep[2]); + edep->data->frag_id = get_int64(&ep[2+8]); + ep += 16; + } #undef CHKSIZE #define CHKSIZE(SZ) \ - do { if ((SZ) > edep->ext_endp - ep) ERTS_EXT_HDR_FAIL; } while(0) + do { if ((SZ) > edep->data->ext_endp - ep) goto bad_hdr; } while(0) CHKSIZE(1+1+1); ep += 2; no_atoms = (int) get_int8(ep); if (no_atoms < 0 || ERTS_ATOM_CACHE_SIZE < no_atoms) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; ep++; if (no_atoms) { int long_atoms = 0; @@ -776,18 +946,18 @@ erts_prepare_dist_ext(ErtsDistExternal *edep, /* atom already cached */ cix += (int) get_int8(ep); if (cix >= ERTS_ATOM_CACHE_SIZE) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; ep++; atom = cache->in_arr[cix]; if (!is_atom(atom)) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; edep->attab.atom[tix] = atom; } else { /* new cached atom */ cix += (int) get_int8(ep); if (cix >= ERTS_ATOM_CACHE_SIZE) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; ep++; if (long_atoms) { CHKSIZE(2); @@ -802,12 +972,10 @@ erts_prepare_dist_ext(ErtsDistExternal *edep, CHKSIZE(len); atom = erts_atom_put((byte *) ep, len, - (utf8_atoms - ? ERTS_ATOM_ENC_UTF8 - : ERTS_ATOM_ENC_LATIN1), + ERTS_ATOM_ENC_UTF8, 0); if (is_non_value(atom)) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; ep += len; cache->in_arr[cix] = atom; edep->attab.atom[tix] = atom; @@ -824,25 +992,24 @@ erts_prepare_dist_ext(ErtsDistExternal *edep, #endif } } - edep->extp = ep; + edep->data->extp = ep; #ifdef ERTS_DEBUG_USE_DIST_SEP if (*ep != VERSION_MAGIC) - ERTS_EXT_HDR_FAIL; + goto bad_hdr; #endif } #ifdef ERTS_DEBUG_USE_DIST_SEP if (*ep != VERSION_MAGIC) - ERTS_EXT_FAIL; + goto fail; #endif - return 0; + erts_de_runlock(dep); + + return ERTS_PREP_DIST_EXT_SUCCESS; #undef CHKSIZE -#undef ERTS_EXT_FAIL -#undef ERTS_EXT_HDR_FAIL - bad_hdr: - if (dep) { + bad_hdr: { erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf(); erts_dsprintf(dsbufp, "%T got a corrupted distribution header from %T " @@ -850,15 +1017,16 @@ erts_prepare_dist_ext(ErtsDistExternal *edep, erts_this_node->sysname, edep->dep->sysname, dist_entry_channel_no(edep->dep)); - for (ep = ext; ep < edep->ext_endp; ep++) + for (ep = ext; ep < edep->data->ext_endp; ep++) erts_dsprintf(dsbufp, ep != ext ? ",%b8u" : "<<%b8u", *ep); erts_dsprintf(dsbufp, ">>"); erts_send_warning_to_logger_nogl(dsbufp); } - fail: - if (dep) - erts_kill_dist_connection(dep, dep->connection_id); - return -1; + fail: { + erts_de_runlock(dep); + erts_kill_dist_connection(dep, conn_id); + } + return ERTS_PREP_DIST_EXT_FAILED; } static void @@ -874,9 +1042,9 @@ bad_dist_ext(ErtsDistExternal *edep) erts_this_node->sysname, dep->sysname, dist_entry_channel_no(dep)); - for (ep = edep->extp; ep < edep->ext_endp; ep++) + for (ep = edep->data->extp; ep < edep->data->ext_endp; ep++) erts_dsprintf(dsbufp, - ep != edep->extp ? ",%b8u" : "<<...,%b8u", + ep != edep->data->extp ? ",%b8u" : "<<...,%b8u", *ep); erts_dsprintf(dsbufp, ">>\n"); erts_dsprintf(dsbufp, "ATOM_CACHE_REF translations: "); @@ -889,35 +1057,37 @@ bad_dist_ext(ErtsDistExternal *edep) erts_dsprintf(dsbufp, ", %d=%T", i, edep->attab.atom[i]); } erts_send_warning_to_logger_nogl(dsbufp); - erts_kill_dist_connection(dep, ERTS_DIST_EXT_CON_ID(edep)); + erts_kill_dist_connection(dep, edep->connection_id); } } Sint -erts_decode_dist_ext_size(ErtsDistExternal *edep) +erts_decode_dist_ext_size(ErtsDistExternal *edep, int kill_connection) { Sint res; byte *ep; - if (edep->extp >= edep->ext_endp) + + if (edep->data->extp >= edep->data->ext_endp) goto fail; #ifndef ERTS_DEBUG_USE_DIST_SEP if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) { - if (*edep->extp == VERSION_MAGIC) + if (*edep->data->extp == VERSION_MAGIC) goto fail; - ep = edep->extp; + ep = edep->data->extp; } else #endif { - if (*edep->extp != VERSION_MAGIC) + if (*edep->data->extp != VERSION_MAGIC) goto fail; - ep = edep->extp+1; + ep = edep->data->extp+1; } - res = decoded_size(ep, edep->ext_endp, 0, NULL); + res = decoded_size(ep, edep->data->ext_endp, 0, NULL); if (res >= 0) return res; fail: - bad_dist_ext(edep); + if (kill_connection) + bad_dist_ext(edep); return -1; } @@ -943,12 +1113,15 @@ Sint erts_decode_ext_size_ets(byte *ext, Uint size) */ Eterm erts_decode_dist_ext(ErtsHeapFactory* factory, - ErtsDistExternal *edep) + ErtsDistExternal *edep, + int kill_connection) { Eterm obj; - byte* ep = edep->extp; + byte* ep; - if (ep >= edep->ext_endp) + ep = edep->data->extp; + + if (ep >= edep->data->ext_endp) goto error; #ifndef ERTS_DEBUG_USE_DIST_SEP if (edep->flags & ERTS_DIST_EXT_DFLAG_HDR) { @@ -966,14 +1139,15 @@ erts_decode_dist_ext(ErtsHeapFactory* factory, if (!ep) goto error; - edep->extp = ep; + edep->data->extp = ep; return obj; error: erts_factory_undo(factory); - bad_dist_ext(edep); + if (kill_connection) + bad_dist_ext(edep); return THE_NON_VALUE; } @@ -1018,6 +1192,7 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2) Eterm res; Sint hsz; ErtsDistExternal ede; + ErtsDistExternalData ede_data; Eterm *tp; Eterm real_bin; Uint offset; @@ -1030,7 +1205,8 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2) ede.flags = ERTS_DIST_EXT_ATOM_TRANS_TAB; ede.dep = NULL; ede.heap_size = -1; - + ede.data = &ede_data; + if (is_not_tuple(BIF_ARG_1)) goto badarg; tp = tuple_val(BIF_ARG_1); @@ -1055,15 +1231,15 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2) if (bitsize != 0) goto badarg; - ede.extp = binary_bytes(real_bin)+offset; - ede.ext_endp = ede.extp + size; + ede.data->extp = binary_bytes(real_bin)+offset; + ede.data->ext_endp = ede.data->extp + size; - hsz = erts_decode_dist_ext_size(&ede); + hsz = erts_decode_dist_ext_size(&ede, 1); if (hsz < 0) goto badarg; erts_factory_proc_prealloc_init(&factory, BIF_P, hsz); - res = erts_decode_dist_ext(&factory, &ede); + res = erts_decode_dist_ext(&factory, &ede, 1); erts_factory_close(&factory); if (is_value(res)) @@ -1078,9 +1254,22 @@ static BIF_RETTYPE term_to_binary_trap_1(BIF_ALIST_1) { Eterm *tp = tuple_val(BIF_ARG_1); Eterm Term = tp[1]; - Eterm bt = tp[2]; + Eterm Opts = tp[2]; + Eterm bt = tp[3]; Binary *bin = erts_magic_ref2bin(bt); - Eterm res = erts_term_to_binary_int(BIF_P, Term, 0, 0,bin); + Eterm res = erts_term_to_binary_int(BIF_P, Term, Opts, 0, 0,bin); + if (is_non_value(res)) { + if (erts_set_gc_state(BIF_P, 1) + || MSO(BIF_P).overhead > BIN_VHEAP_SZ(BIF_P)) { + ERTS_VBUMP_ALL_REDS(BIF_P); + } + if (Opts == am_undefined) + ERTS_BIF_ERROR_TRAPPED1(BIF_P, SYSTEM_LIMIT, + bif_export[BIF_term_to_binary_1], Term); + else + ERTS_BIF_ERROR_TRAPPED2(BIF_P, SYSTEM_LIMIT, + bif_export[BIF_term_to_binary_2], Term, Opts); + } if (is_tuple(res)) { ASSERT(BIF_P->flags & F_DISABLE_GC); BIF_TRAP1(&term_to_binary_trap_export,BIF_P,res); @@ -1097,7 +1286,12 @@ HIPE_WRAPPER_BIF_DISABLE_GC(term_to_binary, 1) BIF_RETTYPE term_to_binary_1(BIF_ALIST_1) { - Eterm res = erts_term_to_binary_int(BIF_P, BIF_ARG_1, 0, TERM_TO_BINARY_DFLAGS, NULL); + Eterm res = erts_term_to_binary_int(BIF_P, BIF_ARG_1, am_undefined, + 0, TERM_TO_BINARY_DFLAGS, NULL); + if (is_non_value(res)) { + ASSERT(!(BIF_P->flags & F_DISABLE_GC)); + BIF_ERROR(BIF_P, SYSTEM_LIMIT); + } if (is_tuple(res)) { erts_set_gc_state(BIF_P, 0); BIF_TRAP1(&term_to_binary_trap_export,BIF_P,res); @@ -1156,7 +1350,12 @@ BIF_RETTYPE term_to_binary_2(BIF_ALIST_2) goto error; } - res = erts_term_to_binary_int(p, Term, level, flags, NULL); + res = erts_term_to_binary_int(p, Term, BIF_ARG_2, + level, flags, NULL); + if (is_non_value(res)) { + ASSERT(!(BIF_P->flags & F_DISABLE_GC)); + BIF_ERROR(BIF_P, SYSTEM_LIMIT); + } if (is_tuple(res)) { erts_set_gc_state(p, 0); BIF_TRAP1(&term_to_binary_trap_export,BIF_P,res); @@ -1214,7 +1413,8 @@ typedef struct B2TContext_t { ErtsBinary2TermState b2ts; Uint32 flags; SWord reds; - Eterm trap_bin; + Uint used_bytes; /* In: boolean, Out: bytes */ + Eterm trap_bin; /* THE_NON_VALUE if not exported */ Export *bif; Eterm arg[2]; enum B2TState state; @@ -1308,6 +1508,11 @@ binary2term_prepare(ErtsBinary2TermState *state, byte *data, Sint data_size, ctx->u.uc.dbytes = state->extp; ctx->u.uc.dleft = dest_len; + if (ctx->used_bytes) { + ASSERT(ctx->used_bytes == 1); + /* to be subtracted by stream.avail_in when done */ + ctx->used_bytes = data_size; + } ctx->state = B2TUncompressChunk; *ctxp = ctx; } @@ -1410,13 +1615,15 @@ static int b2t_context_destructor(Binary *context_bin) return 1; } +static BIF_RETTYPE binary_to_term_int(Process*, Eterm bin, B2TContext*); + + static BIF_RETTYPE binary_to_term_trap_1(BIF_ALIST_1) { Binary *context_bin = erts_magic_ref2bin(BIF_ARG_1); ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(context_bin) == b2t_context_destructor); - return binary_to_term_int(BIF_P, 0, THE_NON_VALUE, context_bin, NULL, - THE_NON_VALUE, THE_NON_VALUE); + return binary_to_term_int(BIF_P, THE_NON_VALUE, ERTS_MAGIC_BIN_DATA(context_bin)); } @@ -1442,6 +1649,8 @@ static B2TContext* b2t_export_context(Process* p, B2TContext* src) b2t_context_destructor); B2TContext* ctx = ERTS_MAGIC_BIN_DATA(context_b); Eterm* hp; + + ASSERT(is_non_value(src->trap_bin)); sys_memcpy(ctx, src, sizeof(B2TContext)); if (ctx->state >= B2TDecode && ctx->u.dc.next == &src->u.dc.res) { ctx->u.dc.next = &ctx->u.dc.res; @@ -1451,8 +1660,7 @@ static B2TContext* b2t_export_context(Process* p, B2TContext* src) return ctx; } -static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binary* context_b, - Export *bif_init, Eterm arg0, Eterm arg1) +static BIF_RETTYPE binary_to_term_int(Process* p, Eterm bin, B2TContext *ctx) { BIF_RETTYPE ret_val; #ifdef EXTREME_B2T_TRAPPING @@ -1460,25 +1668,17 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binar #else SWord initial_reds = (Uint)(ERTS_BIF_REDS_LEFT(p) * B2T_BYTES_PER_REDUCTION); #endif - B2TContext c_buff; - B2TContext *ctx; int is_first_call; - if (context_b == NULL) { + if (is_value(bin)) { /* Setup enough to get started */ is_first_call = 1; - ctx = &c_buff; ctx->state = B2TPrepare; ctx->aligned_alloc = NULL; - ctx->flags = flags; - ctx->bif = bif_init; - ctx->arg[0] = arg0; - ctx->arg[1] = arg1; - IF_DEBUG(ctx->trap_bin = THE_NON_VALUE;) } else { - is_first_call = 0; - ctx = ERTS_MAGIC_BIN_DATA(context_b); + ASSERT(is_value(ctx->trap_bin)); ASSERT(ctx->state != B2TPrepare); + is_first_call = 0; } ctx->reds = initial_reds; @@ -1522,6 +1722,10 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binar && zret == Z_STREAM_END && ctx->u.uc.dleft == 0) { ctx->reds -= chunk; + if (ctx->used_bytes) { + ASSERT(ctx->used_bytes > 5 + ctx->u.uc.stream.avail_in); + ctx->used_bytes -= ctx->u.uc.stream.avail_in; + } ctx->state = B2TSizeInit; } else { @@ -1540,11 +1744,11 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binar break; case B2TDecodeInit: - if (ctx == &c_buff && ctx->b2ts.extsize > ctx->reds) { + if (is_non_value(ctx->trap_bin) && ctx->b2ts.extsize > ctx->reds) { /* dec_term will maybe trap, allocate space for magic bin before result term to make it easy to trim with HRelease. */ - ctx = b2t_export_context(p, &c_buff); + ctx = b2t_export_context(p, ctx); } ctx->u.dc.ep = ctx->b2ts.extp; ctx->u.dc.res = (Eterm) (UWord) NULL; @@ -1587,6 +1791,25 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binar return ret_val; case B2TDone: + if (ctx->used_bytes) { + Eterm *hp; + Eterm used; + if (!ctx->b2ts.exttmp) { + ASSERT(ctx->used_bytes == 1); + ctx->used_bytes = (ctx->u.dc.ep - ctx->b2ts.extp + +1); /* VERSION_MAGIC */ + } + if (IS_USMALL(0, ctx->used_bytes)) { + hp = erts_produce_heap(&ctx->u.dc.factory, 3, 0); + used = make_small(ctx->used_bytes); + } + else { + hp = erts_produce_heap(&ctx->u.dc.factory, 3+BIG_UINT_HEAP_SIZE, 0); + used = uint_to_big(ctx->used_bytes, hp); + hp += BIG_UINT_HEAP_SIZE; + } + ctx->u.dc.res = TUPLE2(hp, ctx->u.dc.res, used); + } b2t_destroy_context(ctx); if (ctx->u.dc.factory.hp > ctx->u.dc.factory.hp_end) { @@ -1607,11 +1830,10 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Uint32 flags, Eterm bin, Binar } }while (ctx->reds > 0 || ctx->state >= B2TDone); - if (ctx == &c_buff) { - ASSERT(ctx->trap_bin == THE_NON_VALUE); - ctx = b2t_export_context(p, &c_buff); + if (is_non_value(ctx->trap_bin)) { + ctx = b2t_export_context(p, ctx); + ASSERT(is_value(ctx->trap_bin)); } - ASSERT(ctx->trap_bin != THE_NON_VALUE); if (is_first_call) { erts_set_gc_state(p, 0); @@ -1628,23 +1850,35 @@ HIPE_WRAPPER_BIF_DISABLE_GC(binary_to_term, 1) BIF_RETTYPE binary_to_term_1(BIF_ALIST_1) { - return binary_to_term_int(BIF_P, 0, BIF_ARG_1, NULL, bif_export[BIF_binary_to_term_1], - BIF_ARG_1, THE_NON_VALUE); + B2TContext ctx; + + ctx.flags = 0; + ctx.used_bytes = 0; + ctx.trap_bin = THE_NON_VALUE; + ctx.bif = bif_export[BIF_binary_to_term_1]; + ctx.arg[0] = BIF_ARG_1; + ctx.arg[1] = THE_NON_VALUE; + return binary_to_term_int(BIF_P, BIF_ARG_1, &ctx); } HIPE_WRAPPER_BIF_DISABLE_GC(binary_to_term, 2) BIF_RETTYPE binary_to_term_2(BIF_ALIST_2) { + B2TContext ctx; Eterm opts; Eterm opt; - Uint32 flags = 0; + ctx.flags = 0; + ctx.used_bytes = 0; opts = BIF_ARG_2; while (is_list(opts)) { opt = CAR(list_val(opts)); if (opt == am_safe) { - flags |= ERTS_DIST_EXT_BTT_SAFE; + ctx.flags |= ERTS_DIST_EXT_BTT_SAFE; + } + else if (opt == am_used) { + ctx.used_bytes = 1; } else { goto error; @@ -1655,8 +1889,11 @@ BIF_RETTYPE binary_to_term_2(BIF_ALIST_2) if (is_not_nil(opts)) goto error; - return binary_to_term_int(BIF_P, flags, BIF_ARG_1, NULL, bif_export[BIF_binary_to_term_2], - BIF_ARG_1, BIF_ARG_2); + ctx.trap_bin = THE_NON_VALUE; + ctx.bif = bif_export[BIF_binary_to_term_2]; + ctx.arg[0] = BIF_ARG_1; + ctx.arg[1] = BIF_ARG_2; + return binary_to_term_int(BIF_P, BIF_ARG_1, &ctx); error: BIF_ERROR(BIF_P, BADARG); @@ -1667,8 +1904,17 @@ external_size_1(BIF_ALIST_1) { Process* p = BIF_P; Eterm Term = BIF_ARG_1; + Uint size; + + switch (erts_encode_ext_size(Term, &size)) { + case ERTS_EXT_SZ_SYSTEM_LIMIT: + BIF_ERROR(BIF_P, SYSTEM_LIMIT); + case ERTS_EXT_SZ_YIELD: + ERTS_INTERNAL_ERROR("Unexpected yield"); + case ERTS_EXT_SZ_OK: + break; + } - Uint size = erts_encode_ext_size(Term); if (IS_USMALL(0, size)) { BIF_RET(make_small(size)); } else { @@ -1711,7 +1957,15 @@ external_size_2(BIF_ALIST_2) goto error; } - size = erts_encode_ext_size_2(BIF_ARG_1, flags); + switch (erts_encode_ext_size_2(BIF_ARG_1, flags, &size)) { + case ERTS_EXT_SZ_SYSTEM_LIMIT: + BIF_ERROR(BIF_P, SYSTEM_LIMIT); + case ERTS_EXT_SZ_YIELD: + ERTS_INTERNAL_ERROR("Unexpected yield"); + case ERTS_EXT_SZ_OK: + break; + } + if (IS_USMALL(0, size)) { BIF_RET(make_small(size)); } else { @@ -1799,7 +2053,15 @@ erts_term_to_binary_simple(Process* p, Eterm Term, Uint size, int level, Uint fl Eterm erts_term_to_binary(Process* p, Eterm Term, int level, Uint flags) { Uint size; - size = encode_size_struct2(NULL, Term, flags) + 1 /* VERSION_MAGIC */; + switch (encode_size_struct_int(NULL, NULL, Term, flags, NULL, &size)) { + case ERTS_EXT_SZ_SYSTEM_LIMIT: + return THE_NON_VALUE; + case ERTS_EXT_SZ_YIELD: + ERTS_INTERNAL_ERROR("Unexpected yield"); + case ERTS_EXT_SZ_OK: + break; + } + size++; /* VERSION_MAGIC */; return erts_term_to_binary_simple(p, Term, size, level, flags); } @@ -1849,8 +2111,8 @@ static int ttb_context_destructor(Binary *context_bin) return 1; } -static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint flags, - Binary *context_b) +static Eterm erts_term_to_binary_int(Process* p, Eterm Term, Eterm opts, int level, + Uint flags, Binary *context_b) { Eterm *hp; Eterm res; @@ -1868,17 +2130,17 @@ static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint fla do { \ if (context_b == NULL) { \ context_b = erts_create_magic_binary(sizeof(TTBContext), \ - ttb_context_destructor); \ + ttb_context_destructor);\ context = ERTS_MAGIC_BIN_DATA(context_b); \ - memcpy(context,&c_buff,sizeof(TTBContext)); \ + sys_memcpy(context,&c_buff,sizeof(TTBContext)); \ } \ } while (0) #define RETURN_STATE() \ do { \ - hp = HAlloc(p, ERTS_MAGIC_REF_THING_SIZE+3); \ + hp = HAlloc(p, ERTS_MAGIC_REF_THING_SIZE + 1 + 3); \ c_term = erts_mk_magic_ref(&hp, &MSO(p), context_b); \ - res = TUPLE2(hp, Term, c_term); \ + res = TUPLE3(hp, Term, opts, c_term); \ BUMP_ALL_REDS(p); \ return res; \ } while (0); @@ -1904,11 +2166,17 @@ static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint fla int level; Uint flags; /* Try for fast path */ - if (encode_size_struct_int(&context->s.sc, NULL, Term, - context->s.sc.flags, &reds, &size) < 0) { + switch (encode_size_struct_int(&context->s.sc, NULL, Term, + context->s.sc.flags, &reds, &size)) { + case ERTS_EXT_SZ_SYSTEM_LIMIT: + BUMP_REDS(p, (initial_reds - reds) / TERM_TO_BINARY_LOOP_FACTOR); + return THE_NON_VALUE; + case ERTS_EXT_SZ_YIELD: EXPORT_CONTEXT(); /* Same state */ RETURN_STATE(); + case ERTS_EXT_SZ_OK: + break; } ++size; /* VERSION_MAGIC */ /* Move these to next state */ @@ -1949,23 +2217,14 @@ static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint fla level = context->s.ec.level; BUMP_REDS(p, (initial_reds - reds) / TERM_TO_BINARY_LOOP_FACTOR); if (level == 0 || real_size < 6) { /* We are done */ - ProcBin* pb; return_normal: context->s.ec.result_bin = NULL; context->alive = 0; - pb = (ProcBin *) HAlloc(p, PROC_BIN_SIZE); - pb->thing_word = HEADER_PROC_BIN; - pb->size = real_size; - pb->next = MSO(p).first; - MSO(p).first = (struct erl_off_heap_header*)pb; - pb->val = result_bin; - pb->bytes = (byte*) result_bin->orig_bytes; - pb->flags = 0; - OH_OVERHEAD(&(MSO(p)), pb->size / sizeof(Eterm)); if (context_b && erts_refc_read(&context_b->intern.refc,0) == 0) { erts_bin_free(context_b); } - return make_binary(pb); + return erts_build_proc_bin(&MSO(p), HAlloc(p, PROC_BIN_SIZE), + result_bin); } /* Continue with compression... */ /* To make absolutely sure that zlib does not barf on a reallocated context, @@ -2022,16 +2281,7 @@ static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint fla result_bin = erts_bin_realloc(context->s.cc.destination_bin, context->s.cc.dest_len+6); context->s.cc.destination_bin = NULL; - pb = (ProcBin *) HAlloc(p, PROC_BIN_SIZE); - pb->thing_word = HEADER_PROC_BIN; - pb->size = context->s.cc.dest_len+6; - pb->next = MSO(p).first; - MSO(p).first = (struct erl_off_heap_header*)pb; - pb->val = result_bin; ASSERT(erts_refc_read(&result_bin->intern.refc, 1)); - pb->bytes = (byte*) result_bin->orig_bytes; - pb->flags = 0; - OH_OVERHEAD(&(MSO(p)), pb->size / sizeof(Eterm)); erts_bin_free(context->s.cc.result_bin); context->s.cc.result_bin = NULL; context->alive = 0; @@ -2039,7 +2289,9 @@ static Eterm erts_term_to_binary_int(Process* p, Eterm Term, int level, Uint fla if (context_b && erts_refc_read(&context_b->intern.refc,0) == 0) { erts_bin_free(context_b); } - return make_binary(pb); + return erts_build_proc_bin(&MSO(p), + HAlloc(p, PROC_BIN_SIZE), + result_bin); } default: /* Compression error, revert to uncompressed binary (still in context) */ @@ -2093,7 +2345,7 @@ enc_atom(ErtsAtomCacheMap *acmp, Eterm atom, byte *ep, Uint32 dflags) { int iix; int len; - int utf8_atoms = (int) (dflags & DFLAG_UTF8_ATOMS); + const int utf8_atoms = (int) (dflags & DFLAG_UTF8_ATOMS); ASSERT(is_atom(atom)); @@ -2286,7 +2538,7 @@ dec_atom(ErtsDistExternal *edep, byte* ep, Eterm* objp) return ep; } -static ERTS_INLINE ErlNode* dec_get_node(Eterm sysname, Uint32 creation) +static ERTS_INLINE ErlNode* dec_get_node(Eterm sysname, Uint32 creation, Eterm book) { if (sysname == INTERNAL_LOCAL_SYSNAME) /* && DFLAG_INTERNAL_TAGS */ return erts_this_node; @@ -2295,7 +2547,7 @@ static ERTS_INLINE ErlNode* dec_get_node(Eterm sysname, Uint32 creation) && (creation == erts_this_node->creation || creation == ORIG_CREATION)) return erts_this_node; - return erts_find_or_insert_node(sysname,creation); + return erts_find_or_insert_node(sysname,creation,book); } static byte* @@ -2341,7 +2593,7 @@ dec_pid(ErtsDistExternal *edep, ErtsHeapFactory* factory, byte* ep, * We are careful to create the node entry only after all * validity tests are done. */ - node = dec_get_node(sysname, cre); + node = dec_get_node(sysname, cre, make_boxed(factory->hp)); if(node == erts_this_node) { *objp = make_internal_pid(data); @@ -2419,11 +2671,21 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, { Eterm* cons = list_val(obj); Eterm tl; + Uint len_cnt = WSTACK_POP(s); obj = CAR(cons); tl = CDR(cons); - WSTACK_PUSH2(s, (is_list(tl) ? ENC_ONE_CONS : ENC_TERM), - tl); + if (is_list(tl)) { + len_cnt++; + WSTACK_PUSH3(s, len_cnt, ENC_ONE_CONS, tl); + } + else { + byte* list_lenp = (byte*) WSTACK_POP(s); + ASSERT(list_lenp[-1] == LIST_EXT); + put_int32(len_cnt, list_lenp); + + WSTACK_PUSH2(s, ENC_TERM, tl); + } } break; case ENC_PATCH_FUN_SIZE: @@ -2488,8 +2750,6 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, break; } - L_jump_start: - if (ctx && --r <= 0) { *reds = 0; ctx->obj = obj; @@ -2497,6 +2757,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, WSTACK_SAVE(s, &ctx->wstack); return -1; } + + L_jump_start: switch(tag_val_def(obj)) { case NIL_DEF: *ep++ = NIL_EXT; @@ -2627,10 +2889,7 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, } case LIST_DEF: { - int is_str; - - i = is_external_string(obj, &is_str); - if (is_str) { + if (is_external_string(obj, &i)) { *ep++ = STRING_EXT; put_int16(i, ep); ep += 2; @@ -2639,9 +2898,12 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, *ep++ = unsigned_val(CAR(cons)); obj = CDR(cons); } + r -= i; } else { + r -= i/2; *ep++ = LIST_EXT; - put_int32(i, ep); + /* Patch list length when we find end of list */ + WSTACK_PUSH2(s, (UWord)ep, 1); ep += 4; goto encode_one_cons; } @@ -2806,6 +3068,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, ep[j] = 0; /* Zero unused bits at end of binary */ data_dst = ep; ep += j + 1; + if (ctx) + ctx->hopefull_flags |= DFLAG_BIT_BINARIES; } else { /* * Bit-level binary, but the receiver doesn't support it. @@ -2841,6 +3105,8 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, ep = enc_atom(acmp, exp->info.mfa.function, ep, dflags); ep = enc_term(acmp, make_small(exp->info.mfa.arity), ep, dflags, off_heap); + if (ctx) + ctx->hopefull_flags |= DFLAG_EXPORT_PTR_TAG; } else { /* Tag, arity */ *ep++ = SMALL_TUPLE_EXT; @@ -2861,62 +3127,27 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, ErlFunThing* funp = (ErlFunThing *) fun_val(obj); int ei; - if ((dflags & DFLAG_NEW_FUN_TAGS) != 0) { - *ep++ = NEW_FUN_EXT; - WSTACK_PUSH2(s, ENC_PATCH_FUN_SIZE, - (UWord) ep); /* Position for patching in size */ - ep += 4; - *ep = funp->arity; - ep += 1; - sys_memcpy(ep, funp->fe->uniq, 16); - ep += 16; - put_int32(funp->fe->index, ep); - ep += 4; - put_int32(funp->num_free, ep); - ep += 4; - ep = enc_atom(acmp, funp->fe->module, ep, dflags); - ep = enc_term(acmp, make_small(funp->fe->old_index), ep, dflags, off_heap); - ep = enc_term(acmp, make_small(funp->fe->old_uniq), ep, dflags, off_heap); - ep = enc_pid(acmp, funp->creator, ep, dflags); - } else { - /* - * Communicating with an obsolete erl_interface or - * jinterface node. Convert the fun to a tuple to - * avoid crasching. - */ - - /* Tag, arity */ - *ep++ = SMALL_TUPLE_EXT; - put_int8(5, ep); - ep += 1; - - /* 'fun' */ - ep = enc_atom(acmp, am_fun, ep, dflags); - - /* Module name */ - ep = enc_atom(acmp, funp->fe->module, ep, dflags); - - /* Index, Uniq */ - *ep++ = INTEGER_EXT; - put_int32(funp->fe->old_index, ep); - ep += 4; - *ep++ = INTEGER_EXT; - put_int32(funp->fe->old_uniq, ep); - ep += 4; - - /* Environment sub-tuple arity */ - ASSERT(funp->num_free < MAX_ARG); - *ep++ = SMALL_TUPLE_EXT; - put_int8(funp->num_free, ep); - ep += 1; - } - for (ei = funp->num_free-1; ei > 0; ei--) { + ASSERT(dflags & DFLAG_NEW_FUN_TAGS); + *ep++ = NEW_FUN_EXT; + WSTACK_PUSH2(s, ENC_PATCH_FUN_SIZE, + (UWord) ep); /* Position for patching in size */ + ep += 4; + *ep = funp->arity; + ep += 1; + sys_memcpy(ep, funp->fe->uniq, 16); + ep += 16; + put_int32(funp->fe->index, ep); + ep += 4; + put_int32(funp->num_free, ep); + ep += 4; + ep = enc_atom(acmp, funp->fe->module, ep, dflags); + ep = enc_term(acmp, make_small(funp->fe->old_index), ep, dflags, off_heap); + ep = enc_term(acmp, make_small(funp->fe->old_uniq), ep, dflags, off_heap); + ep = enc_pid(acmp, funp->creator, ep, dflags); + + for (ei = funp->num_free-1; ei >= 0; ei--) { WSTACK_PUSH2(s, ENC_TERM, (UWord) funp->env[ei]); } - if (funp->num_free != 0) { - obj = funp->env[0]; - goto L_jump_start; - } } break; } @@ -2930,9 +3161,13 @@ enc_term_int(TTBEncodeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, byte* ep, return 0; } +/** @brief Is it a list of bytes not longer than MAX_STRING_LEN? + * @param lenp out: string length or number of list cells traversed + * @return true/false + */ static -Uint -is_external_string(Eterm list, int* p_is_string) +int +is_external_string(Eterm list, Uint* lenp) { Uint len = 0; @@ -2944,29 +3179,15 @@ is_external_string(Eterm list, int* p_is_string) Eterm* consp = list_val(list); Eterm hd = CAR(consp); - if (!is_byte(hd)) { - break; + if (!is_byte(hd) || ++len > MAX_STRING_LEN) { + *lenp = len; + return 0; } - len++; list = CDR(consp); } - /* - * If we have reached the end of the list, and we have - * not exceeded the maximum length of a string, this - * is a string. - */ - *p_is_string = is_nil(list) && len < MAX_STRING_LEN; - - /* - * Continue to calculate the length. - */ - while (is_list(list)) { - Eterm* consp = list_val(list); - len++; - list = CDR(consp); - } - return len; + *lenp = len; + return is_nil(list); } @@ -3109,7 +3330,7 @@ dec_term(ErtsDistExternal *edep, #if defined(ARCH_64) *objp = make_small(sn); #else - if (MY_IS_SSMALL(sn)) { + if (IS_SSMALL(sn)) { *objp = make_small(sn); } else { *objp = small_to_big(sn, hp); @@ -3254,6 +3475,7 @@ dec_term_atom_common: n--; if (ctx) { if (reds < n) { + ASSERT(reds > 0); ctx->state = B2TDecodeList; ctx->u.dc.remaining_n = n - reds; n = reds; @@ -3365,7 +3587,7 @@ dec_term_atom_common: cre = get_int32(ep); ep += 4; } - node = dec_get_node(sysname, cre); + node = dec_get_node(sysname, cre, make_boxed(hp)); if(node == erts_this_node) { *objp = make_internal_port(num); } @@ -3436,7 +3658,7 @@ dec_term_atom_common: cre = get_int32(ep); ep += 4; - r0 = get_int32(ep); /* allow full word */ + r0 = get_int32(ep); ep += 4; ref_ext_common: { @@ -3445,8 +3667,15 @@ dec_term_atom_common: if (ref_words > ERTS_MAX_REF_NUMBERS) goto error; - node = dec_get_node(sysname, cre); + node = dec_get_node(sysname, cre, make_boxed(hp)); if(node == erts_this_node) { + if (r0 >= MAX_REFERENCE) { + /* + * Must reject local refs with more than 18 bits + * in first word as magic ref table relies on it. + */ + goto error; + } rtp = (ErtsORefThing *) hp; ref_num = &rtp->num[0]; @@ -3535,18 +3764,9 @@ dec_term_atom_common: *objp = make_binary(hb); } else { Binary* dbin = erts_bin_nrml_alloc(n); - ProcBin* pb; - pb = (ProcBin *) hp; + + *objp = erts_build_proc_bin(factory->off_heap, hp, dbin); hp += PROC_BIN_SIZE; - pb->thing_word = HEADER_PROC_BIN; - pb->size = n; - pb->next = factory->off_heap->first; - factory->off_heap->first = (struct erl_off_heap_header*)pb; - OH_OVERHEAD(factory->off_heap, pb->size / sizeof(Eterm)); - pb->val = dbin; - pb->bytes = (byte*) dbin->orig_bytes; - pb->flags = 0; - *objp = make_binary(pb); if (ctx) { int n_limit = reds * B2T_MEMCPY_FACTOR; if (n > n_limit) { @@ -3586,18 +3806,9 @@ dec_term_atom_common: ep += n; } else { Binary* dbin = erts_bin_nrml_alloc(n); - ProcBin* pb; + Uint n_copy = n; - pb = (ProcBin *) hp; - pb->thing_word = HEADER_PROC_BIN; - pb->size = n; - pb->next = factory->off_heap->first; - factory->off_heap->first = (struct erl_off_heap_header*)pb; - OH_OVERHEAD(factory->off_heap, pb->size / sizeof(Eterm)); - pb->val = dbin; - pb->bytes = (byte*) dbin->orig_bytes; - pb->flags = 0; - bin = make_binary(pb); + bin = erts_build_proc_bin(factory->off_heap, hp, dbin); hp += PROC_BIN_SIZE; if (ctx) { int n_limit = reds * B2T_MEMCPY_FACTOR; @@ -3605,15 +3816,15 @@ dec_term_atom_common: ctx->state = B2TDecodeBinary; ctx->u.dc.remaining_n = n - n_limit; ctx->u.dc.remaining_bytes = dbin->orig_bytes + n_limit; - n = n_limit; + n_copy = n_limit; reds = 0; } - else + else { reds -= n / B2T_MEMCPY_FACTOR; + } } - sys_memcpy(dbin->orig_bytes, ep, n); - ep += n; - n = pb->size; + sys_memcpy(dbin->orig_bytes, ep, n_copy); + ep += n_copy; } if (bitsize == 8 || n == 0) { @@ -3994,6 +4205,7 @@ dec_term_atom_common: if (ctx) { ctx->state = B2TDone; ctx->reds = reds; + ctx->u.dc.ep = ep; } return ep; @@ -4026,13 +4238,21 @@ error_hamt: to a sequence of bytes N.B. That this must agree with to_external2() above!!! (except for cached atoms) */ -static Uint encode_size_struct2(ErtsAtomCacheMap *acmp, Eterm obj, unsigned dflags) { - Uint res; - (void) encode_size_struct_int(NULL, acmp, obj, dflags, NULL, &res); - return res; +static Uint encode_size_struct2(ErtsAtomCacheMap *acmp, + Eterm obj, + unsigned dflags) { + Uint size; + ErtsExtSzRes res = encode_size_struct_int(NULL, acmp, obj, + dflags, NULL, &size); + /* + * encode_size_struct2() only allowed when + * we know the result will always be OK! + */ + ASSERT(res == ERTS_EXT_SZ_OK); (void) res; + return (Uint) size; } -static int +static ErtsExtSzRes encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, unsigned dflags, Sint *reds, Uint *res) { @@ -4060,12 +4280,12 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, for (;;) { ASSERT(!is_header(obj)); - if (ctx && --r == 0) { - *reds = r; + if (ctx && --r <= 0) { + *reds = 0; ctx->obj = obj; ctx->result = result; WSTACK_SAVE(s, &ctx->wstack); - return -1; + return ERTS_EXT_SZ_YIELD; } switch (tag_val_def(obj)) { case NIL_DEF: @@ -4151,8 +4371,10 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, result += (1 + encode_size_struct2(acmp, port_node_name(obj), dflags) + 4 + 1); break; - case LIST_DEF: - if ((m = is_string(obj)) && (m < MAX_STRING_LEN)) { + case LIST_DEF: { + int is_str = is_external_string(obj, &m); + r -= m/2; + if (is_str) { result += m + 2 + 1; } else { result += 5; @@ -4161,6 +4383,7 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, continue; /* big loop */ } break; + } case TUPLE_DEF: { Eterm* ptr = tuple_val(obj); @@ -4238,11 +4461,26 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, result += 32; /* Yes, including the tag */ } break; - case BINARY_DEF: - if (dflags & DFLAG_INTERNAL_TAGS) { + case BINARY_DEF: { + ProcBin* pb = (ProcBin*) binary_val(obj); + Uint tot_bytes = pb->size; + if (!(dflags & DFLAG_INTERNAL_TAGS)) { +#ifdef ARCH_64 + if (tot_bytes >= (Uint) 0xffffffff) { + if (pb->thing_word == HEADER_SUB_BIN) { + ErlSubBin* sub = (ErlSubBin*) pb; + tot_bytes += (sub->bitoffs + sub->bitsize+ 7) / 8; + } + if (tot_bytes > (Uint) 0xffffffff) { + WSTACK_DESTROY(s); + return ERTS_EXT_SZ_SYSTEM_LIMIT; + } + } +#endif + } + else { ProcBin* pb = (ProcBin*) binary_val(obj); Uint sub_extra = 0; - Uint tot_bytes = pb->size; if (pb->thing_word == HEADER_SUB_BIN) { ErlSubBin* sub = (ErlSubBin*) pb; pb = (ProcBin*) binary_val(sub->orig); @@ -4259,28 +4497,17 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, result += 1 + 4 + binary_size(obj) + 5; /* For unaligned binary */ break; + } case FUN_DEF: { ErlFunThing* funp = (ErlFunThing *) fun_val(obj); - if ((dflags & DFLAG_NEW_FUN_TAGS) != 0) { - result += 20+1+1+4; /* New ID + Tag */ - result += 4; /* Length field (number of free variables */ - result += encode_size_struct2(acmp, funp->creator, dflags); - result += encode_size_struct2(acmp, funp->fe->module, dflags); - result += 2 * (1+4); /* Index, Uniq */ - } else { - /* - * Size when fun is mapped to a tuple. - */ - result += 1 + 1; /* Tuple tag, arity */ - result += 1 + 1 + 2 + - atom_tab(atom_val(am_fun))->len; /* 'fun' */ - result += 1 + 1 + 2 + - atom_tab(atom_val(funp->fe->module))->len; /* Module name */ - result += 2 * (1 + 4); /* Index + Uniq */ - result += 1 + (funp->num_free < 0x100 ? 1 : 4); - } + ASSERT(dflags & DFLAG_NEW_FUN_TAGS); + result += 20+1+1+4; /* New ID + Tag */ + result += 4; /* Length field (number of free variables */ + result += encode_size_struct2(acmp, funp->creator, dflags); + result += encode_size_struct2(acmp, funp->fe->module, dflags); + result += 2 * (1+4); /* Index, Uniq */ if (funp->num_free > 1) { WSTACK_PUSH2(s, (UWord) (funp->env + 1), (UWord) TERM_ARRAY_OP(funp->num_free-1)); @@ -4303,7 +4530,7 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, break; default: - erts_exit(ERTS_ERROR_EXIT,"Internal data structure error (in encode_size_struct2)%x\n", + erts_exit(ERTS_ERROR_EXIT,"Internal data structure error (in encode_size_struct_int) %x\n", obj); } @@ -4314,7 +4541,7 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, if (is_header(obj)) { switch (obj) { - case LIST_TAIL_OP: + case LIST_TAIL_OP: obj = (Eterm) WSTACK_POP(s); if (is_list(obj)) { Eterm* cons = list_val(obj); @@ -4340,10 +4567,10 @@ encode_size_struct_int(TTBSizeContext* ctx, ErtsAtomCacheMap *acmp, Eterm obj, WSTACK_DESTROY(s); if (ctx) { ASSERT(ctx->wstack.wstart == NULL); - *reds = r; + *reds = r < 0 ? 0 : r; } *res = result; - return 0; + return ERTS_EXT_SZ_OK; } @@ -4368,7 +4595,7 @@ decoded_size(byte *ep, byte* endp, int internal_tags, B2TContext* ctx) } } else - reds = 0; /* not used but compiler warns anyway */ + ERTS_UNDEF(reds, 0); heap_size = 0; terms = 1; @@ -4678,3 +4905,182 @@ error: #undef SKIP2 #undef CHKSIZE } + + +struct transcode_context { + enum { + TRANSCODE_DEC_MSG_SIZE, + TRANSCODE_DEC_MSG, + TRANSCODE_ENC_CTL, + TRANSCODE_ENC_MSG + }state; + Eterm ctl_term; + Eterm* ctl_heap; + ErtsHeapFactory ctl_factory; + Eterm* msg_heap; + B2TContext b2t; + TTBEncodeContext ttb; +#ifdef DEBUG + ErtsDistOutputBuf* dbg_ob; +#endif +}; + +void transcode_free_ctx(DistEntry* dep) +{ + struct transcode_context* ctx = dep->transcode_ctx; + + erts_factory_close(&ctx->ctl_factory); + erts_free(ERTS_ALC_T_DIST_TRANSCODE, ctx->ctl_heap); + + if (ctx->msg_heap) { + erts_factory_close(&ctx->b2t.u.dc.factory); + erts_free(ERTS_ALC_T_DIST_TRANSCODE, ctx->msg_heap); + } + erts_free(ERTS_ALC_T_DIST_TRANSCODE, ctx); + dep->transcode_ctx = NULL; +} + +Sint transcode_dist_obuf(ErtsDistOutputBuf* ob, + DistEntry* dep, + Uint32 dflags, + Sint reds) +{ + Sint hsz; + byte* decp; + const int have_msg = !!ob->msg_start; + int i; + struct transcode_context* ctx = dep->transcode_ctx; + + if (!ctx) { /* first call for 'ob' */ + ASSERT(!(ob->hopefull_flags & ~(Uint)(DFLAG_BIT_BINARIES | + DFLAG_EXPORT_PTR_TAG))); + if (~dflags & ob->hopefull_flags) { + /* + * Receiver does not support bitstrings and/or export funs + * and output buffer contains such message tags (hopefull_flags). + * Must transcode control and message terms to use tuple fallbacks. + */ + ctx = erts_alloc(ERTS_ALC_T_DIST_TRANSCODE, sizeof(struct transcode_context)); + dep->transcode_ctx = ctx; + #ifdef DEBUG + ctx->dbg_ob = ob; + #endif + + hsz = decoded_size(ob->extp, ob->ext_endp, 0, NULL); + ctx->ctl_heap = erts_alloc(ERTS_ALC_T_DIST_TRANSCODE, hsz*sizeof(Eterm)); + erts_factory_tmp_init(&ctx->ctl_factory, ctx->ctl_heap, hsz, ERTS_ALC_T_DIST_TRANSCODE); + ctx->msg_heap = NULL; + + decp = dec_term(NULL, &ctx->ctl_factory, ob->extp, &ctx->ctl_term, NULL); + if (have_msg) { + ASSERT(decp == ob->msg_start); (void)decp; + ctx->b2t.u.sc.ep = NULL; + ctx->b2t.state = B2TSize; + ctx->b2t.aligned_alloc = NULL; + ctx->b2t.b2ts.exttmp = 0; + ctx->state = TRANSCODE_DEC_MSG_SIZE; + } + else { + ASSERT(decp == ob->ext_endp); + ctx->state = TRANSCODE_ENC_CTL; + } + } + else if (!(dflags & DFLAG_DIST_HDR_ATOM_CACHE)) { + /* + * No need for full transcoding, but primitive receiver (erl_/jinterface) + * expects VERSION_MAGIC before both control and message terms. + */ + if (ob->msg_start) { + Sint ctl_bytes = ob->msg_start - ob->extp; + ASSERT(ob->extp < ob->msg_start && ob->msg_start < ob->ext_endp); + /* Move control term back 1 byte to make room */ + sys_memmove(ob->extp-1, ob->extp, ctl_bytes); + *--(ob->msg_start) = VERSION_MAGIC; + --(ob->extp); + reds -= ctl_bytes / (B2T_BYTES_PER_REDUCTION * B2T_MEMCPY_FACTOR); + } + *--(ob->extp) = VERSION_MAGIC; + goto done; + } + else + goto done; + } + else { /* continue after yield */ + ASSERT(ctx->dbg_ob == ob); + } + ctx->b2t.reds = reds * B2T_BYTES_PER_REDUCTION; + + switch (ctx->state) { + case TRANSCODE_DEC_MSG_SIZE: + hsz = decoded_size(ob->msg_start, ob->ext_endp, 0, &ctx->b2t); + if (ctx->b2t.state == B2TSize) { + return -1; + } + ASSERT(ctx->b2t.state == B2TDecodeInit); + ctx->msg_heap = erts_alloc(ERTS_ALC_T_DIST_TRANSCODE, hsz*sizeof(Eterm)); + ctx->b2t.u.dc.ep = ob->msg_start; + ctx->b2t.u.dc.res = (Eterm) NULL; + ctx->b2t.u.dc.next = &ctx->b2t.u.dc.res; + erts_factory_tmp_init(&ctx->b2t.u.dc.factory, + ctx->msg_heap, hsz, ERTS_ALC_T_DIST_TRANSCODE); + ctx->b2t.u.dc.flat_maps.wstart = NULL; + ctx->b2t.u.dc.hamt_array.pstart = NULL; + ctx->b2t.state = B2TDecode; + + ctx->state = TRANSCODE_DEC_MSG; + case TRANSCODE_DEC_MSG: + if (ctx->b2t.reds <= 0) + ctx->b2t.reds = 1; + decp = dec_term(NULL, NULL, NULL, NULL, &ctx->b2t); + if (ctx->b2t.state < B2TDone) { + return -1; + } + ASSERT(ctx->b2t.state == B2TDone); + ASSERT(decp && decp <= ob->ext_endp); + reds = ctx->b2t.reds / B2T_BYTES_PER_REDUCTION; + b2t_destroy_context(&ctx->b2t); + + ctx->state = TRANSCODE_ENC_CTL; + case TRANSCODE_ENC_CTL: + if (!(dflags & DFLAG_DIST_HDR_ATOM_CACHE)) { + ASSERT(!(dflags & DFLAG_NO_MAGIC)); + ob->extp -= 2; /* VERSION_MAGIC x 2 */ + } + ob->ext_endp = ob->extp; + i = erts_encode_dist_ext(ctx->ctl_term, &ob->ext_endp, dflags, + NULL, NULL, NULL); + ASSERT(i == 0); (void)i; + ASSERT(ob->ext_endp <= ob->alloc_endp); + + if (!have_msg) { + break; + } + ob->msg_start = ob->ext_endp; + ctx->ttb.wstack.wstart = NULL; + ctx->ttb.flags = dflags; + ctx->ttb.hopefull_flags = 0; + ctx->ttb.level = 0; + + ctx->state = TRANSCODE_ENC_MSG; + case TRANSCODE_ENC_MSG: + reds *= TERM_TO_BINARY_LOOP_FACTOR; + if (erts_encode_dist_ext(ctx->b2t.u.dc.res, &ob->ext_endp, dflags, NULL, + &ctx->ttb, &reds)) { + return -1; + } + reds /= TERM_TO_BINARY_LOOP_FACTOR; + + ASSERT(ob->ext_endp <= ob->alloc_endp); + ASSERT(!ctx->ttb.hopefull_flags); + } + transcode_free_ctx(dep); + +done: + if (!(dflags & DFLAG_DIST_HDR_ATOM_CACHE)) + *--(ob->extp) = PASS_THROUGH; + + if (reds < 0) + reds = 0; + + return reds; +} |