erts: Add erlang:iolist_to_iovec

OTP-14520

1 files changed, 500 insertions, 0 deletions

diff --git a/erts/emulator/beam/erl_io_queue.c b/erts/emulator/beam/erl_io_queue.c
index d072376d1c..301b55b315 100644
--- a/erts/emulator/beam/erl_io_queue.c
+++ b/erts/emulator/beam/erl_io_queue.c
@@ -24,9 +24,24 @@
 
 #include "sys.h"
 #include "global.h"
+
+#define ERL_WANT_HIPE_BIF_WRAPPER__
+#include "bif.h"
+#undef ERL_WANT_HIPE_BIF_WRAPPER__
+
 #include "erl_bits.h"
 #include "erl_io_queue.h"
 
+#ifndef MAX
+#define MAX(X, Y) ((X) > (Y) ? (X) : (Y))
+#endif
+
+#ifndef MIN
+#define MIN(X, Y) ((X) < (Y) ? (X) : (Y))
+#endif
+
+#define IOL2V_SMALL_BIN_LIMIT (ERL_ONHEAP_BIN_LIMIT * 4)
+
 static void free_binary(ErtsIOQBinary *b, int driver);
 static ErtsIOQBinary *alloc_binary(Uint size, char *source, void **iov_base, int driver);
 
@@ -736,3 +751,488 @@ erts_ioq_iolist_vec_len(Eterm obj, int* vsize, Uint* csize,
     DESTROY_ESTACK(s);
     return 1;
 }
+
+typedef struct {
+    Eterm result_head;
+    Eterm result_tail;
+    Eterm input_list;
+
+    UWord acc_size;
+    Binary *acc;
+
+    /* We yield after copying this many bytes into the accumulator (Minus
+     * eating a few on consing etc). Large binaries will only count to the
+     * extent their split (if any) resulted in a copy op. */
+    UWord bytereds_available;
+    UWord bytereds_spent;
+
+    Process *process;
+    ErtsEStack estack;
+
+    Eterm magic_reference;
+} iol2v_state_t;
+
+static int iol2v_state_destructor(Binary *data) {
+    iol2v_state_t *state = ERTS_MAGIC_BIN_UNALIGNED_DATA(data);
+
+    DESTROY_SAVED_ESTACK(&state->estack);
+
+    if (state->acc != NULL) {
+        erts_bin_free(state->acc);
+    }
+
+    return 1;
+}
+
+static void iol2v_init(iol2v_state_t *state, Process *process, Eterm input) {
+    state->process = process;
+
+    state->result_head = NIL;
+    state->result_tail = NIL;
+    state->input_list = input;
+
+    state->magic_reference = NIL;
+    state->acc_size = 0;
+    state->acc = NULL;
+
+    CLEAR_SAVED_ESTACK(&state->estack);
+}
+
+static Eterm iol2v_make_sub_bin(iol2v_state_t *state, Eterm bin_term,
+        UWord offset, UWord size) {
+    Uint byte_offset, bit_offset, bit_size;
+    ErlSubBin *sb;
+    Eterm orig_pb_term;
+
+    sb = (ErlSubBin*)HAlloc(state->process, ERL_SUB_BIN_SIZE);
+
+    ERTS_GET_REAL_BIN(bin_term, orig_pb_term,
+        byte_offset, bit_offset, bit_size);
+
+    (void)bit_offset;
+    (void)bit_size;
+
+    sb->thing_word = HEADER_SUB_BIN;
+    sb->bitsize = 0;
+    sb->bitoffs = 0;
+    sb->orig = orig_pb_term;
+    sb->is_writable = 0;
+
+    sb->offs = byte_offset + offset;
+    sb->size = size;
+
+    return make_binary(sb);
+}
+
+static Eterm iol2v_promote_acc(iol2v_state_t *state) {
+    ProcBin *pb;
+
+    state->acc = erts_bin_realloc(state->acc, state->acc_size);
+
+    pb = (ProcBin*)HAlloc(state->process, PROC_BIN_SIZE);
+    pb->thing_word = HEADER_PROC_BIN;
+    pb->size = state->acc_size;
+    pb->val = state->acc;
+    pb->bytes = (byte*)(state->acc)->orig_bytes;
+    pb->flags = 0;
+    pb->next = MSO(state->process).first;
+    OH_OVERHEAD(&(MSO(state->process)), pb->size / sizeof(Eterm));
+    MSO(state->process).first = (struct erl_off_heap_header*)pb;
+
+    state->acc_size = 0;
+    state->acc = NULL;
+
+    return make_binary(pb);
+}
+
+/* Destructively enqueues a term to the result list, saving us the hassle of
+ * having to reverse it later. This is safe since GC is disabled and we never
+ * leak the unfinished term to the outside. */
+static void iol2v_enqueue_result(iol2v_state_t *state, Eterm term) {
+    Eterm prev_tail;
+    Eterm *hp;
+
+    prev_tail = state->result_tail;
+
+    hp = HAlloc(state->process, 2);
+    state->result_tail = CONS(hp, term, NIL);
+
+    if(prev_tail != NIL) {
+        Eterm *prev_cell = list_val(prev_tail);
+        CDR(prev_cell) = state->result_tail;
+    } else {
+        state->result_head = state->result_tail;
+    }
+
+    state->bytereds_spent += 1;
+}
+
+#ifndef DEBUG
+    #define ACC_REALLOCATION_LIMIT (IOL2V_SMALL_BIN_LIMIT * 32)
+#else
+    #define ACC_REALLOCATION_LIMIT (IOL2V_SMALL_BIN_LIMIT * 4)
+#endif
+
+static void iol2v_expand_acc(iol2v_state_t *state, UWord extra) {
+    UWord required_bytes, acc_alloc_size;
+
+    ERTS_CT_ASSERT(ERTS_UWORD_MAX > ACC_REALLOCATION_LIMIT / 2);
+    ASSERT(extra >= 1);
+
+    acc_alloc_size = state->acc != NULL ? (state->acc)->orig_size : 0;
+    required_bytes = state->acc_size + extra;
+
+    if (state->acc == NULL) {
+        UWord new_size = MAX(required_bytes, IOL2V_SMALL_BIN_LIMIT);
+
+        state->acc = erts_bin_nrml_alloc(new_size);
+    } else if (required_bytes > acc_alloc_size) {
+        Binary *prev_acc;
+        UWord new_size;
+
+        if (acc_alloc_size >= ACC_REALLOCATION_LIMIT) {
+            /* We skip reallocating once we hit a certain point; it often
+             * results in extra copying and we're very likely to overallocate
+             * on anything other than absurdly long byte/heapbin sequences. */
+            iol2v_enqueue_result(state, iol2v_promote_acc(state));
+            iol2v_expand_acc(state, extra);
+            return;
+        }
+
+        new_size = MAX(required_bytes, acc_alloc_size * 2);
+        prev_acc = state->acc;
+
+        state->acc = erts_bin_realloc(prev_acc, new_size);
+
+        if (prev_acc != state->acc) {
+            state->bytereds_spent += state->acc_size;
+        }
+    }
+
+    state->bytereds_spent += extra;
+}
+
+static int iol2v_append_byte_seq(iol2v_state_t *state, Eterm seq_start, Eterm *seq_end) {
+    Eterm lookahead, iterator;
+    Uint observed_bits;
+    SWord seq_length;
+    char *acc_data;
+
+    lookahead = seq_start;
+    seq_length = 0;
+
+    ASSERT(state->bytereds_available > state->bytereds_spent);
+
+    while (is_list(lookahead)) {
+        Eterm *cell = list_val(lookahead);
+
+        if (!is_small(CAR(cell))) {
+            break;
+        }
+
+        if (seq_length * 2 >= (state->bytereds_available - state->bytereds_spent)) {
+            break;
+        }
+
+        lookahead = CDR(cell);
+        seq_length += 1;
+    }
+
+    ASSERT(seq_length >= 1);
+
+    iol2v_expand_acc(state, seq_length);
+
+    /* Bump a few extra reductions to account for list traversal. */
+    state->bytereds_spent += seq_length;
+
+    acc_data = &(state->acc)->orig_bytes[state->acc_size];
+    state->acc_size += seq_length;
+
+    iterator = seq_start;
+    observed_bits = 0;
+
+    while (iterator != lookahead) {
+        Eterm *cell;
+        Uint byte;
+
+        cell = list_val(iterator);
+        iterator = CDR(cell);
+
+        byte = unsigned_val(CAR(cell));
+        observed_bits |= byte;
+
+        ASSERT(acc_data < &(state->acc)->orig_bytes[state->acc_size]);
+        *(acc_data++) = byte;
+    }
+
+    if (observed_bits > UCHAR_MAX) {
+        return 0;
+    }
+
+    ASSERT(acc_data == &(state->acc)->orig_bytes[state->acc_size]);
+    *seq_end = iterator;
+
+    return 1;
+}
+
+static int iol2v_append_binary(iol2v_state_t *state, Eterm bin_term) {
+    int is_acc_small, is_bin_small;
+    UWord combined_size;
+    UWord binary_size;
+
+    Uint byte_offset, bit_offset, bit_size;
+    Eterm *parent_header;
+    Eterm parent_binary;
+    byte *binary_data;
+
+    ASSERT(state->bytereds_available > state->bytereds_spent);
+
+    ERTS_GET_REAL_BIN(bin_term, parent_binary, byte_offset, bit_offset, bit_size);
+    parent_header = binary_val(parent_binary);
+    binary_size = binary_size(bin_term);
+
+    if (bit_offset != 0 || bit_size != 0) {
+        return 0;
+    } else if (binary_size == 0) {
+        state->bytereds_spent += 1;
+        return 1;
+    }
+
+    is_acc_small = state->acc_size < IOL2V_SMALL_BIN_LIMIT;
+    is_bin_small = binary_size < IOL2V_SMALL_BIN_LIMIT;
+    combined_size = binary_size + state->acc_size;
+
+    if (thing_subtag(*parent_header) == REFC_BINARY_SUBTAG) {
+        ProcBin *pb = (ProcBin*)parent_header;
+
+        if (pb->flags) {
+            erts_emasculate_writable_binary(pb);
+        }
+
+        binary_data = pb->bytes;
+    } else {
+        ErlHeapBin *hb = (ErlHeapBin*)parent_header;
+
+        ASSERT(thing_subtag(*parent_header) == HEAP_BINARY_SUBTAG);
+        ASSERT(is_bin_small);
+
+        binary_data = &((unsigned char*)&hb->data)[byte_offset];
+    }
+
+    if (!is_bin_small && (state->acc_size == 0 || !is_acc_small)) {
+        /* Avoid combining if we encounter an acceptably large binary while the
+         * accumulator is either empty or large enough to be returned on its
+         * own. */
+        if (state->acc_size != 0) {
+            iol2v_enqueue_result(state, iol2v_promote_acc(state));
+        }
+
+        iol2v_enqueue_result(state, bin_term);
+    } else if (is_bin_small || combined_size < (IOL2V_SMALL_BIN_LIMIT * 2)) {
+        /* If the candidate is small or we can't split the combination in two,
+         * then just copy it into the accumulator. */
+        iol2v_expand_acc(state, binary_size);
+
+        sys_memcpy(&(state->acc)->orig_bytes[state->acc_size],
+            binary_data, binary_size);
+
+        state->acc_size += binary_size;
+    } else {
+        /* Otherwise, append enough data for the accumulator to be valid, and
+         * then return the rest as a sub-binary. */
+        UWord spill = IOL2V_SMALL_BIN_LIMIT - state->acc_size;
+        Eterm binary_tail;
+
+        iol2v_expand_acc(state, spill);
+
+        sys_memcpy(&(state->acc)->orig_bytes[state->acc_size],
+            binary_data, spill);
+
+        state->acc_size += spill;
+
+        binary_tail = iol2v_make_sub_bin(state, bin_term, spill,
+            binary_size - spill);
+
+        iol2v_enqueue_result(state, iol2v_promote_acc(state));
+        iol2v_enqueue_result(state, binary_tail);
+    }
+
+    return 1;
+}
+
+static BIF_RETTYPE iol2v_yield(iol2v_state_t *state) {
+    if (is_nil(state->magic_reference)) {
+        iol2v_state_t *boxed_state;
+        Binary *magic_binary;
+        Eterm *hp;
+
+        magic_binary = erts_create_magic_binary_x(sizeof(*state),
+            &iol2v_state_destructor, ERTS_ALC_T_BINARY, 1);
+
+        boxed_state = ERTS_MAGIC_BIN_UNALIGNED_DATA(magic_binary);
+        sys_memcpy(boxed_state, state, sizeof(*state));
+
+        hp = HAlloc(boxed_state->process, ERTS_MAGIC_REF_THING_SIZE);
+        boxed_state->magic_reference =
+            erts_mk_magic_ref(&hp, &MSO(boxed_state->process), magic_binary);
+
+        state = boxed_state;
+    }
+
+    ERTS_BIF_YIELD1(bif_export[BIF_iolist_to_iovec_1],
+        state->process, state->magic_reference);
+}
+
+static BIF_RETTYPE iol2v_continue(iol2v_state_t *state) {
+    Eterm iterator;
+
+    DECLARE_ESTACK(s);
+    ESTACK_CHANGE_ALLOCATOR(s, ERTS_ALC_T_SAVED_ESTACK);
+
+    state->bytereds_available =
+        ERTS_BIF_REDS_LEFT(state->process) * IOL2V_SMALL_BIN_LIMIT;
+    state->bytereds_spent = 0;
+
+    if (state->estack.start) {
+        ESTACK_RESTORE(s, &state->estack);
+    }
+
+    iterator = state->input_list;
+
+    for(;;) {
+        if (state->bytereds_spent >= state->bytereds_available) {
+            ESTACK_SAVE(s, &state->estack);
+            state->input_list = iterator;
+
+            return iol2v_yield(state);
+        }
+
+        while (is_list(iterator)) {
+            Eterm *cell;
+            Eterm head;
+
+            cell = list_val(iterator);
+            head = CAR(cell);
+
+            if (is_binary(head)) {
+                if (!iol2v_append_binary(state, head)) {
+                    goto l_badarg;
+                }
+
+                iterator = CDR(cell);
+            } else if (is_small(head)) {
+                Eterm seq_end;
+
+                if (!iol2v_append_byte_seq(state, iterator, &seq_end)) {
+                    goto l_badarg;
+                }
+
+                iterator = seq_end;
+            } else if (is_list(head) || is_nil(head)) {
+                Eterm tail = CDR(cell);
+
+                if (!is_nil(tail)) {
+                    ESTACK_PUSH(s, tail);
+                }
+
+                state->bytereds_spent += 1;
+                iterator = head;
+            } else {
+                goto l_badarg;
+            }
+
+            if (state->bytereds_spent >= state->bytereds_available) {
+                ESTACK_SAVE(s, &state->estack);
+                state->input_list = iterator;
+
+                return iol2v_yield(state);
+            }
+        }
+
+        if (is_binary(iterator)) {
+            if (!iol2v_append_binary(state, iterator)) {
+                goto l_badarg;
+            }
+        } else if (!is_nil(iterator)) {
+            goto l_badarg;
+        }
+
+        if(ESTACK_ISEMPTY(s)) {
+            break;
+        }
+
+        iterator = ESTACK_POP(s);
+    }
+
+    if (state->acc_size != 0) {
+        iol2v_enqueue_result(state, iol2v_promote_acc(state));
+    }
+
+    BUMP_REDS(state->process, state->bytereds_spent / IOL2V_SMALL_BIN_LIMIT);
+
+    CLEAR_SAVED_ESTACK(&state->estack);
+    DESTROY_ESTACK(s);
+
+    BIF_RET(state->result_head);
+
+l_badarg:
+    CLEAR_SAVED_ESTACK(&state->estack);
+    DESTROY_ESTACK(s);
+
+    if (state->acc != NULL) {
+        erts_bin_free(state->acc);
+        state->acc = NULL;
+    }
+
+    BIF_ERROR(state->process, BADARG);
+}
+
+HIPE_WRAPPER_BIF_DISABLE_GC(iolist_to_iovec, 1)
+
+BIF_RETTYPE iolist_to_iovec_1(BIF_ALIST_1) {
+    BIF_RETTYPE result;
+
+    if (is_nil(BIF_ARG_1)) {
+        BIF_RET(NIL);
+    } else if (is_binary(BIF_ARG_1)) {
+        if (binary_size(BIF_ARG_1) != 0) {
+            Eterm *hp = HAlloc(BIF_P, 2);
+
+            BIF_RET(CONS(hp, BIF_ARG_1, NIL));
+        } else {
+            BIF_RET(NIL);
+        }
+    } else if (is_internal_magic_ref(BIF_ARG_1)) {
+        iol2v_state_t *state;
+        Binary *magic;
+
+        magic = erts_magic_ref2bin(BIF_ARG_1);
+
+        if (ERTS_MAGIC_BIN_DESTRUCTOR(magic) != &iol2v_state_destructor) {
+            ASSERT(!(BIF_P->flags & F_DISABLE_GC));
+            BIF_ERROR(BIF_P, BADARG);
+        }
+
+        ASSERT(BIF_P->flags & F_DISABLE_GC);
+
+        state = ERTS_MAGIC_BIN_UNALIGNED_DATA(magic);
+        result = iol2v_continue(state);
+    } else if (!is_list(BIF_ARG_1)) {
+        ASSERT(!(BIF_P->flags & F_DISABLE_GC));
+        BIF_ERROR(BIF_P, BADARG);
+    } else {
+        iol2v_state_t state;
+
+        iol2v_init(&state, BIF_P, BIF_ARG_1);
+
+        erts_set_gc_state(BIF_P, 0);
+
+        result = iol2v_continue(&state);
+    }
+
+    if (result != THE_NON_VALUE || BIF_P->freason != TRAP) {
+        erts_set_gc_state(BIF_P, 1);
+    }
+
+    BIF_RET(result);
+}