diff options
Diffstat (limited to 'erts/emulator/beam/erl_io_queue.c')
-rw-r--r-- | erts/emulator/beam/erl_io_queue.c | 738 |
1 files changed, 738 insertions, 0 deletions
diff --git a/erts/emulator/beam/erl_io_queue.c b/erts/emulator/beam/erl_io_queue.c new file mode 100644 index 0000000000..d072376d1c --- /dev/null +++ b/erts/emulator/beam/erl_io_queue.c @@ -0,0 +1,738 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 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% + */ + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include "sys.h" +#include "global.h" +#include "erl_bits.h" +#include "erl_io_queue.h" + +static void free_binary(ErtsIOQBinary *b, int driver); +static ErtsIOQBinary *alloc_binary(Uint size, char *source, void **iov_base, int driver); + +void erts_ioq_init(ErtsIOQueue *q, ErtsAlcType_t alct, int driver) +{ + + ERTS_CT_ASSERT(offsetof(ErlNifIOVec,flags) == sizeof(ErtsIOVecCommon)); + ERTS_CT_ASSERT(sizeof(ErlIOVec) == sizeof(ErtsIOVecCommon)); + ERTS_CT_ASSERT(sizeof(size_t) == sizeof(ErlDrvSizeT)); + ERTS_CT_ASSERT(sizeof(size_t) == sizeof(Uint)); + + q->alct = alct; + q->driver = driver; + q->size = 0; + q->v_head = q->v_tail = q->v_start = q->v_small; + q->v_end = q->v_small + ERTS_SMALL_IO_QUEUE; + q->b_head = q->b_tail = q->b_start = q->b_small; + q->b_end = q->b_small + ERTS_SMALL_IO_QUEUE; +} + +void erts_ioq_clear(ErtsIOQueue *q) +{ + ErtsIOQBinary** binp = q->b_head; + int driver = q->driver; + + if (q->v_start != q->v_small) + erts_free(q->alct, (void *) q->v_start); + + while(binp < q->b_tail) { + if (*binp != NULL) + free_binary(*binp, driver); + binp++; + } + if (q->b_start != q->b_small) + erts_free(q->alct, (void *) q->b_start); + q->v_start = q->v_end = q->v_head = q->v_tail = NULL; + q->b_start = q->b_end = q->b_head = q->b_tail = NULL; + q->size = 0; +} + +static void free_binary(ErtsIOQBinary *b, int driver) +{ + if (driver) + driver_free_binary(&b->driver); + else if (erts_refc_dectest(&b->nif.intern.refc, 0) == 0) + erts_bin_free(&b->nif); +} + +static ErtsIOQBinary *alloc_binary(Uint size, char *source, void **iov_base, int driver) +{ + if (driver) { + ErlDrvBinary *bin = driver_alloc_binary(size); + if (!bin) return NULL; + sys_memcpy(bin->orig_bytes, source, size); + *iov_base = bin->orig_bytes; + return (ErtsIOQBinary *)bin; + } else { + /* This clause can be triggered in enif_ioq_enq_binary is used */ + Binary *bin = erts_bin_nrml_alloc(size); + if (!bin) return NULL; + erts_refc_init(&bin->intern.refc, 1); + sys_memcpy(bin->orig_bytes, source, size); + *iov_base = bin->orig_bytes; + return (ErtsIOQBinary *)bin; + } +} + +Uint erts_ioq_size(ErtsIOQueue *q) +{ + return q->size; +} + +/* expand queue to hold n elements in tail or head */ +static int expandq(ErtsIOQueue* q, int n, int tail) +/* tail: 0 if make room in head, make room in tail otherwise */ +{ + int h_sz; /* room before header */ + int t_sz; /* room after tail */ + int q_sz; /* occupied */ + int nvsz; + SysIOVec* niov; + ErtsIOQBinary** nbinv; + + h_sz = q->v_head - q->v_start; + t_sz = q->v_end - q->v_tail; + q_sz = q->v_tail - q->v_head; + + if (tail && (n <= t_sz)) /* do we need to expand tail? */ + return 0; + else if (!tail && (n <= h_sz)) /* do we need to expand head? */ + return 0; + else if (n > (h_sz + t_sz)) { /* need to allocate */ + /* we may get little extra but it ok */ + nvsz = (q->v_end - q->v_start) + n; + + niov = erts_alloc_fnf(q->alct, nvsz * sizeof(SysIOVec)); + if (!niov) + return -1; + nbinv = erts_alloc_fnf(q->alct, nvsz * sizeof(ErtsIOQBinary**)); + if (!nbinv) { + erts_free(q->alct, (void *) niov); + return -1; + } + if (tail) { + sys_memcpy(niov, q->v_head, q_sz*sizeof(SysIOVec)); + if (q->v_start != q->v_small) + erts_free(q->alct, (void *) q->v_start); + q->v_start = niov; + q->v_end = niov + nvsz; + q->v_head = q->v_start; + q->v_tail = q->v_head + q_sz; + + sys_memcpy(nbinv, q->b_head, q_sz*sizeof(ErtsIOQBinary*)); + if (q->b_start != q->b_small) + erts_free(q->alct, (void *) q->b_start); + q->b_start = nbinv; + q->b_end = nbinv + nvsz; + q->b_head = q->b_start; + q->b_tail = q->b_head + q_sz; + } + else { + sys_memcpy(niov+nvsz-q_sz, q->v_head, q_sz*sizeof(SysIOVec)); + if (q->v_start != q->v_small) + erts_free(q->alct, (void *) q->v_start); + q->v_start = niov; + q->v_end = niov + nvsz; + q->v_tail = q->v_end; + q->v_head = q->v_tail - q_sz; + + sys_memcpy(nbinv+nvsz-q_sz, q->b_head, q_sz*sizeof(ErtsIOQBinary*)); + if (q->b_start != q->b_small) + erts_free(q->alct, (void *) q->b_start); + q->b_start = nbinv; + q->b_end = nbinv + nvsz; + q->b_tail = q->b_end; + q->b_head = q->b_tail - q_sz; + } + } + else if (tail) { /* move to beginning to make room in tail */ + sys_memmove(q->v_start, q->v_head, q_sz*sizeof(SysIOVec)); + q->v_head = q->v_start; + q->v_tail = q->v_head + q_sz; + sys_memmove(q->b_start, q->b_head, q_sz*sizeof(ErtsIOQBinary*)); + q->b_head = q->b_start; + q->b_tail = q->b_head + q_sz; + } + else { /* move to end to make room */ + sys_memmove(q->v_end-q_sz, q->v_head, q_sz*sizeof(SysIOVec)); + q->v_tail = q->v_end; + q->v_head = q->v_tail-q_sz; + sys_memmove(q->b_end-q_sz, q->b_head, q_sz*sizeof(ErtsIOQBinary*)); + q->b_tail = q->b_end; + q->b_head = q->b_tail-q_sz; + } + + return 0; +} + +static +int skip(ErtsIOVec* vec, Uint skipbytes, + SysIOVec **iovp, ErtsIOQBinary ***binvp, + Uint *lenp) +{ + int n; + Uint len; + SysIOVec* iov; + ErtsIOQBinary** binv; + + if (vec->common.size <= skipbytes) + return -1; + + iov = vec->common.iov; + binv = vec->common.binv; + n = vec->common.vsize; + /* we use do here to strip iov_len=0 from beginning */ + do { + len = iov->iov_len; + if (len <= skipbytes) { + skipbytes -= len; + iov++; + binv++; + n--; + } + else { + iov->iov_base = ((char *)(iov->iov_base)) + skipbytes; + iov->iov_len -= skipbytes; + skipbytes = 0; + } + } while(skipbytes > 0); + + *binvp = binv; + *iovp = iov; + *lenp = len; + + return n; +} + +/* Put elements from vec at q tail */ +int erts_ioq_enqv(ErtsIOQueue *q, ErtsIOVec *eiov, Uint skipbytes) +{ + int n; + Uint len; + Uint size = eiov->common.size - skipbytes; + SysIOVec *iov; + ErtsIOQBinary** binv; + ErtsIOQBinary* b; + + if (q == NULL) + return -1; + + ASSERT(eiov->common.size >= skipbytes); + if (eiov->common.size <= skipbytes) + return 0; + + n = skip(eiov, skipbytes, &iov, &binv, &len); + + if (n < 0) + return n; + + if (q->v_tail + n >= q->v_end) + if (expandq(q, n, 1)) + return -1; + + /* Queue and reference all binaries (remove zero length items) */ + while(n--) { + if ((len = iov->iov_len) > 0) { + if ((b = *binv) == NULL) { /* special case create binary ! */ + b = alloc_binary(len, iov->iov_base, (void**)&q->v_tail->iov_base, + q->driver); + if (!b) return -1; + *q->b_tail++ = b; + q->v_tail->iov_len = len; + q->v_tail++; + } + else { + if (q->driver) + driver_binary_inc_refc(&b->driver); + else + erts_refc_inc(&b->nif.intern.refc, 1); + *q->b_tail++ = b; + *q->v_tail++ = *iov; + } + } + iov++; + binv++; + } + q->size += size; /* update total size in queue */ + return 0; +} + +/* Put elements from vec at q head */ +int erts_ioq_pushqv(ErtsIOQueue *q, ErtsIOVec* vec, Uint skipbytes) +{ + int n; + Uint len; + Uint size = vec->common.size - skipbytes; + SysIOVec* iov; + ErtsIOQBinary** binv; + ErtsIOQBinary* b; + + if (q == NULL) + return -1; + + ASSERT(vec->common.size >= skipbytes); + if (vec->common.size <= skipbytes) + return 0; + + n = skip(vec, skipbytes, &iov, &binv, &len); + + if (n < 0) + return n; + + if (q->v_head - n < q->v_start) + if (expandq(q, n, 0)) + return -1; + + /* Queue and reference all binaries (remove zero length items) */ + iov += (n-1); /* move to end */ + binv += (n-1); /* move to end */ + while(n--) { + if ((len = iov->iov_len) > 0) { + if ((b = *binv) == NULL) { /* special case create binary ! */ + if (q->driver) { + ErlDrvBinary *bin = driver_alloc_binary(len); + if (!bin) return -1; + sys_memcpy(bin->orig_bytes, iov->iov_base, len); + b = (ErtsIOQBinary *)bin; + q->v_head->iov_base = bin->orig_bytes; + } + *--q->b_head = b; + q->v_head--; + q->v_head->iov_len = len; + } + else { + if (q->driver) + driver_binary_inc_refc(&b->driver); + else + erts_refc_inc(&b->nif.intern.refc, 1); + *--q->b_head = b; + *--q->v_head = *iov; + } + } + iov--; + binv--; + } + q->size += size; /* update total size in queue */ + return 0; +} + + +/* +** Remove size bytes from queue head +** Return number of bytes that remain in queue +*/ +int erts_ioq_deq(ErtsIOQueue *q, Uint size) +{ + Uint len; + + if ((q == NULL) || (q->size < size)) + return -1; + q->size -= size; + while (size > 0) { + ASSERT(q->v_head != q->v_tail); + + len = q->v_head->iov_len; + if (len <= size) { + size -= len; + free_binary(*q->b_head, q->driver); + *q->b_head++ = NULL; + q->v_head++; + } + else { + q->v_head->iov_base = ((char *)(q->v_head->iov_base)) + size; + q->v_head->iov_len -= size; + size = 0; + } + } + + /* restart pointers (optimised for enq) */ + if (q->v_head == q->v_tail) { + q->v_head = q->v_tail = q->v_start; + q->b_head = q->b_tail = q->b_start; + } + return 0; +} + + +Uint erts_ioq_peekqv(ErtsIOQueue *q, ErtsIOVec *ev) { + ASSERT(ev); + + if (! q) { + return (Uint) -1; + } else { + if ((ev->common.vsize = q->v_tail - q->v_head) == 0) { + ev->common.size = 0; + ev->common.iov = NULL; + ev->common.binv = NULL; + } else { + ev->common.size = q->size; + ev->common.iov = q->v_head; + ev->common.binv = q->b_head; + } + return q->size; + } +} + +SysIOVec* erts_ioq_peekq(ErtsIOQueue *q, int* vlenp) /* length of io-vector */ +{ + + if (q == NULL) { + *vlenp = -1; + return NULL; + } + if ((*vlenp = (q->v_tail - q->v_head)) == 0) + return NULL; + return q->v_head; +} + +/* Fills a possibly deep list of chars and binaries into vec +** Small characters are first stored in the buffer buf of length ln +** binaries found are copied and linked into msoh +** Return vector length on succsess, +** -1 on overflow +** -2 on type error +*/ + +static ERTS_INLINE void +io_list_to_vec_set_vec(SysIOVec **iov, ErtsIOQBinary ***binv, + ErtsIOQBinary *bin, byte *ptr, Uint len, + int *vlen) +{ + while (len > MAX_SYSIOVEC_IOVLEN) { + (*iov)->iov_base = ptr; + (*iov)->iov_len = MAX_SYSIOVEC_IOVLEN; + ptr += MAX_SYSIOVEC_IOVLEN; + len -= MAX_SYSIOVEC_IOVLEN; + (*iov)++; + (*vlen)++; + *(*binv)++ = bin; + } + (*iov)->iov_base = ptr; + (*iov)->iov_len = len; + *(*binv)++ = bin; + (*iov)++; + (*vlen)++; +} + +int +erts_ioq_iolist_to_vec(Eterm obj, /* io-list */ + SysIOVec* iov, /* io vector */ + ErtsIOQBinary** binv, /* binary reference vector */ + ErtsIOQBinary* cbin, /* binary to store characters */ + Uint bin_limit, /* small binaries limit */ + int driver) +{ + DECLARE_ESTACK(s); + Eterm* objp; + byte *buf = NULL; + Uint len = 0; + Uint csize = 0; + int vlen = 0; + byte* cptr; + + if (cbin) { + if (driver) { + buf = (byte*)cbin->driver.orig_bytes; + len = cbin->driver.orig_size; + } else { + buf = (byte*)cbin->nif.orig_bytes; + len = cbin->nif.orig_size; + } + } + cptr = buf; + + goto L_jump_start; /* avoid push */ + + while (!ESTACK_ISEMPTY(s)) { + obj = ESTACK_POP(s); + L_jump_start: + if (is_list(obj)) { + L_iter_list: + objp = list_val(obj); + obj = CAR(objp); + if (is_byte(obj)) { + if (len == 0) + goto L_overflow; + *buf++ = unsigned_val(obj); + csize++; + len--; + } else if (is_binary(obj)) { + ESTACK_PUSH(s, CDR(objp)); + goto handle_binary; + } else if (is_list(obj)) { + ESTACK_PUSH(s, CDR(objp)); + goto L_iter_list; /* on head */ + } else if (!is_nil(obj)) { + goto L_type_error; + } + obj = CDR(objp); + if (is_list(obj)) + goto L_iter_list; /* on tail */ + else if (is_binary(obj)) { + goto handle_binary; + } else if (!is_nil(obj)) { + goto L_type_error; + } + } else if (is_binary(obj)) { + Eterm real_bin; + Uint offset; + Eterm* bptr; + Uint size; + int bitoffs; + int bitsize; + + handle_binary: + size = binary_size(obj); + ERTS_GET_REAL_BIN(obj, real_bin, offset, bitoffs, bitsize); + ASSERT(bitsize == 0); + bptr = binary_val(real_bin); + if (*bptr == HEADER_PROC_BIN) { + ProcBin* pb = (ProcBin *) bptr; + if (bitoffs != 0) { + if (len < size) { + goto L_overflow; + } + erts_copy_bits(pb->bytes+offset, bitoffs, 1, + (byte *) buf, 0, 1, size*8); + csize += size; + buf += size; + len -= size; + } else if (bin_limit && size < bin_limit) { + if (len < size) { + goto L_overflow; + } + sys_memcpy(buf, pb->bytes+offset, size); + csize += size; + buf += size; + len -= size; + } else { + ErtsIOQBinary *qbin; + if (csize != 0) { + io_list_to_vec_set_vec(&iov, &binv, cbin, + cptr, csize, &vlen); + cptr = buf; + csize = 0; + } + if (pb->flags) { + erts_emasculate_writable_binary(pb); + } + if (driver) + qbin = (ErtsIOQBinary*)Binary2ErlDrvBinary(pb->val); + else + qbin = (ErtsIOQBinary*)pb->val; + + io_list_to_vec_set_vec( + &iov, &binv, qbin, + pb->bytes+offset, size, &vlen); + } + } else { + ErlHeapBin* hb = (ErlHeapBin *) bptr; + if (len < size) { + goto L_overflow; + } + copy_binary_to_buffer(buf, 0, + ((byte *) hb->data)+offset, bitoffs, + 8*size); + csize += size; + buf += size; + len -= size; + } + } else if (!is_nil(obj)) { + goto L_type_error; + } + } + + if (csize != 0) { + io_list_to_vec_set_vec(&iov, &binv, cbin, cptr, csize, &vlen); + } + + DESTROY_ESTACK(s); + return vlen; + + L_type_error: + DESTROY_ESTACK(s); + return -2; + + L_overflow: + DESTROY_ESTACK(s); + return -1; +} + +static ERTS_INLINE int +io_list_vec_count(Eterm obj, Uint *v_size, + Uint *c_size, Uint *b_size, Uint *in_clist, + Uint *p_v_size, Uint *p_c_size, Uint *p_in_clist, + Uint blimit) +{ + Uint size = binary_size(obj); + Eterm real; + ERTS_DECLARE_DUMMY(Uint offset); + int bitoffs; + int bitsize; + ERTS_GET_REAL_BIN(obj, real, offset, bitoffs, bitsize); + if (bitsize != 0) return 1; + if (thing_subtag(*binary_val(real)) == REFC_BINARY_SUBTAG && + bitoffs == 0) { + *b_size += size; + if (*b_size < size) return 2; + *in_clist = 0; + ++*v_size; + /* If iov_len is smaller then Uint we split the binary into*/ + /* multiple smaller (2GB) elements in the iolist.*/ + *v_size += size / MAX_SYSIOVEC_IOVLEN; + if (size >= blimit) { + *p_in_clist = 0; + ++*p_v_size; + } else { + *p_c_size += size; + if (!*p_in_clist) { + *p_in_clist = 1; + ++*p_v_size; + } + } + } else { + *c_size += size; + if (*c_size < size) return 2; + if (!*in_clist) { + *in_clist = 1; + ++*v_size; + } + *p_c_size += size; + if (!*p_in_clist) { + *p_in_clist = 1; + ++*p_v_size; + } + } + return 0; +} + +#define IO_LIST_VEC_COUNT(obj) \ + do { \ + switch (io_list_vec_count(obj, &v_size, &c_size, \ + &b_size, &in_clist, \ + &p_v_size, &p_c_size, &p_in_clist, \ + blimit)) { \ + case 1: goto L_type_error; \ + case 2: goto L_overflow_error; \ + default: break; \ + } \ + } while(0) + +/* + * Returns 0 if successful and a non-zero value otherwise. + * + * Return values through pointers: + * *vsize - SysIOVec size needed for a writev + * *csize - Number of bytes not in binary (in the common binary) + * *pvsize - SysIOVec size needed if packing small binaries + * *pcsize - Number of bytes in the common binary if packing + * *total_size - Total size of iolist in bytes + */ +int +erts_ioq_iolist_vec_len(Eterm obj, int* vsize, Uint* csize, + Uint* pvsize, Uint* pcsize, + Uint* total_size, Uint blimit) +{ + DECLARE_ESTACK(s); + Eterm* objp; + Uint v_size = 0; + Uint c_size = 0; + Uint b_size = 0; + Uint in_clist = 0; + Uint p_v_size = 0; + Uint p_c_size = 0; + Uint p_in_clist = 0; + Uint total; + + goto L_jump_start; /* avoid a push */ + + while (!ESTACK_ISEMPTY(s)) { + obj = ESTACK_POP(s); + L_jump_start: + if (is_list(obj)) { + L_iter_list: + objp = list_val(obj); + obj = CAR(objp); + + if (is_byte(obj)) { + c_size++; + if (c_size == 0) { + goto L_overflow_error; + } + if (!in_clist) { + in_clist = 1; + v_size++; + } + p_c_size++; + if (!p_in_clist) { + p_in_clist = 1; + p_v_size++; + } + } + else if (is_binary(obj)) { + IO_LIST_VEC_COUNT(obj); + } + else if (is_list(obj)) { + ESTACK_PUSH(s, CDR(objp)); + goto L_iter_list; /* on head */ + } + else if (!is_nil(obj)) { + goto L_type_error; + } + + obj = CDR(objp); + if (is_list(obj)) + goto L_iter_list; /* on tail */ + else if (is_binary(obj)) { /* binary tail is OK */ + IO_LIST_VEC_COUNT(obj); + } + else if (!is_nil(obj)) { + goto L_type_error; + } + } + else if (is_binary(obj)) { + IO_LIST_VEC_COUNT(obj); + } + else if (!is_nil(obj)) { + goto L_type_error; + } + } + + total = c_size + b_size; + if (total < c_size) { + goto L_overflow_error; + } + *total_size = total; + + DESTROY_ESTACK(s); + *vsize = v_size; + *csize = c_size; + *pvsize = p_v_size; + *pcsize = p_c_size; + return 0; + + L_type_error: + L_overflow_error: + DESTROY_ESTACK(s); + return 1; +} |