diff options
Diffstat (limited to 'erts/emulator/beam/io.c')
| -rw-r--r-- | erts/emulator/beam/io.c | 1292 |
1 files changed, 380 insertions, 912 deletions
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index c3eb610fdc..13b8125a99 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -52,6 +52,7 @@ #include "erl_bif_unique.h" #include "erl_hl_timer.h" #include "erl_time.h" +#include "erl_io_queue.h" extern ErlDrvEntry fd_driver_entry; extern ErlDrvEntry vanilla_driver_entry; @@ -108,7 +109,7 @@ static void driver_monitor_unlock_pdl(Port *p); #define ERL_SMALL_IO_BIN_LIMIT (4*ERL_ONHEAP_BIN_LIMIT) #define SMALL_WRITE_VEC 16 -static ERTS_INLINE ErlIOQueue* +static ERTS_INLINE ErlPortIOQueue* drvport2ioq(ErlDrvPort drvport) { Port *prt = erts_thr_drvport2port(drvport, 0); @@ -123,11 +124,11 @@ is_port_ioq_empty(Port *pp) int res; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp)); if (!pp->port_data_lock) - res = (pp->ioq.size == 0); + res = (erts_ioq_size(&pp->ioq) == 0); else { ErlDrvPDL pdl = pp->port_data_lock; erts_mtx_lock(&pdl->mtx); - res = (pp->ioq.size == 0); + res = (erts_ioq_size(&pp->ioq) == 0); erts_mtx_unlock(&pdl->mtx); } return res; @@ -142,14 +143,14 @@ erts_is_port_ioq_empty(Port *pp) Uint erts_port_ioq_size(Port *pp) { - int res; + ErlDrvSizeT res; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp)); if (!pp->port_data_lock) - res = pp->ioq.size; + res = erts_ioq_size(&pp->ioq); else { ErlDrvPDL pdl = pp->port_data_lock; erts_mtx_lock(&pdl->mtx); - res = pp->ioq.size; + res = erts_ioq_size(&pp->ioq); erts_mtx_unlock(&pdl->mtx); } return (Uint) res; @@ -258,14 +259,7 @@ static ERTS_INLINE void port_init_instr(Port *prt #ifdef ERTS_SMP ASSERT(prt->drv_ptr && prt->lock); if (!prt->drv_ptr->lock) { - char *lock_str = "port_lock"; - erts_mtx_init_locked_x(prt->lock, lock_str, id, -#ifdef ERTS_ENABLE_LOCK_COUNT - (erts_lcnt_rt_options & ERTS_LCNT_OPT_PORTLOCK) -#else - 0 -#endif - ); + erts_mtx_init_locked(prt->lock, "port_lock", id, ERTS_LOCK_FLAGS_CATEGORY_IO); } #endif erts_port_task_init_sched(&prt->sched, id); @@ -515,41 +509,17 @@ erts_port_free(Port *prt) */ static void initq(Port* prt) { - ErlIOQueue* q = &prt->ioq; - ERTS_LC_ASSERT(!prt->port_data_lock); - - q->size = 0; - q->v_head = q->v_tail = q->v_start = q->v_small; - q->v_end = q->v_small + SMALL_IO_QUEUE; - q->b_head = q->b_tail = q->b_start = q->b_small; - q->b_end = q->b_small + SMALL_IO_QUEUE; + erts_ioq_init(&prt->ioq, ERTS_ALC_T_IOQ, 1); } static void stopq(Port* prt) { - ErlIOQueue* q; - ErlDrvBinary** binp; if (prt->port_data_lock) driver_pdl_lock(prt->port_data_lock); - q = &prt->ioq; - binp = q->b_head; - - if (q->v_start != q->v_small) - erts_free(ERTS_ALC_T_IOQ, (void *) q->v_start); - - while(binp < q->b_tail) { - if (*binp != NULL) - driver_free_binary(*binp); - binp++; - } - if (q->b_start != q->b_small) - erts_free(ERTS_ALC_T_IOQ, (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; + erts_ioq_clear(&prt->ioq); if (prt->port_data_lock) { driver_pdl_unlock(prt->port_data_lock); @@ -930,311 +900,6 @@ int erts_port_handle_xports(Port *prt) } #endif -/* 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 -*/ - -#ifdef DEBUG -#define MAX_SYSIOVEC_IOVLEN (1ull << (32 - 1)) -#else -#define MAX_SYSIOVEC_IOVLEN (1ull << (sizeof(((SysIOVec*)0)->iov_len) * 8 - 1)) -#endif - -static ERTS_INLINE void -io_list_to_vec_set_vec(SysIOVec **iov, ErlDrvBinary ***binv, - ErlDrvBinary *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)++; -} - -static int -io_list_to_vec(Eterm obj, /* io-list */ - SysIOVec* iov, /* io vector */ - ErlDrvBinary** binv, /* binary reference vector */ - ErlDrvBinary* cbin, /* binary to store characters */ - ErlDrvSizeT bin_limit) /* small binaries limit */ -{ - DECLARE_ESTACK(s); - Eterm* objp; - byte *buf = (byte*)cbin->orig_bytes; - Uint len = cbin->orig_size; - Uint csize = 0; - int vlen = 0; - byte* 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; - ErlDrvSizeT 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 { - 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); - } - io_list_to_vec_set_vec( - &iov, &binv, Binary2ErlDrvBinary(pb->val), - 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; -} - -#define IO_LIST_VEC_COUNT(obj) \ -do { \ - 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) goto L_type_error; \ - if (thing_subtag(*binary_val(_real)) == REFC_BINARY_SUBTAG && \ - _bitoffs == 0) { \ - b_size += _size; \ - if (b_size < _size) goto L_overflow_error; \ - 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 >= ERL_SMALL_IO_BIN_LIMIT) { \ - 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) goto L_overflow_error; \ - if (!in_clist) { \ - in_clist = 1; \ - v_size++; \ - } \ - p_c_size += _size; \ - if (!p_in_clist) { \ - p_in_clist = 1; \ - p_v_size++; \ - } \ - } \ -} 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 - */ - -static int -io_list_vec_len(Eterm obj, int* vsize, Uint* csize, - Uint* pvsize, Uint* pcsize, - ErlDrvSizeT* total_size) -{ - 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 = (ErlDrvSizeT) 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; -} - typedef enum { ERTS_TRY_IMM_DRV_CALL_OK, ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK, @@ -1447,7 +1112,7 @@ finalize_force_imm_drv_call(ErtsTryImmDrvCallState *sp) erts_unblock_fpe(sp->fpe_was_unmasked); } -#define ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE (REF_THING_SIZE + 3) +#define ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE (ERTS_REF_THING_SIZE + 3) static ERTS_INLINE void queue_port_sched_op_reply(Process *rp, @@ -1462,7 +1127,7 @@ queue_port_sched_op_reply(Process *rp, ref= make_internal_ref(hp); write_ref_thing(hp, ref_num[0], ref_num[1], ref_num[2]); - hp += REF_THING_SIZE; + hp += ERTS_REF_THING_SIZE; msg = TUPLE2(hp, ref, msg); @@ -1511,7 +1176,7 @@ port_sched_op_reply(Eterm to, Uint32 *ref_num, Eterm msg, Port* prt) } -ErtsPortOpResult +static ErtsPortOpResult erts_schedule_proc2port_signal(Process *c_p, Port *prt, Eterm caller, @@ -1804,8 +1469,7 @@ cleanup_scheduled_outputv(ErlIOVec *ev, ErlDrvBinary *cbinp) int i; /* Need to free all binaries */ for (i = 1; i < ev->vsize; i++) - if (ev->binv[i]) - driver_free_binary(ev->binv[i]); + driver_free_binary(ev->binv[i]); if (cbinp) driver_free_binary(cbinp); } @@ -1968,21 +1632,19 @@ int erts_port_output_async(Port *prt, Eterm from, Eterm list) { - ErtsPortOpResult res; ErtsProc2PortSigData *sigdp; erts_driver_t *drv = prt->drv_ptr; size_t size; int task_flags; ErtsProc2PortSigCallback port_sig_callback; - ErlDrvBinary *cbin = NULL; - ErlIOVec *evp = NULL; + ErtsIOQBinary *cbin = NULL; + ErtsIOVec *evp = NULL; char *buf = NULL; ErtsPortTaskHandle *ns_pthp; if (drv->outputv) { - ErlIOVec ev; SysIOVec* ivp; - ErlDrvBinary** bvp; + ErtsIOQBinary** bvp; int vsize; Uint csize; Uint pvsize; @@ -1992,91 +1654,63 @@ erts_port_output_async(Port *prt, Eterm from, Eterm list) char *ptr; int i; - Eterm* bptr = NULL; - Uint offset; - - if (is_binary(list)) { - /* We optimize for when we get a procbin without offset */ - Eterm real_bin; - int bitoffs; - int bitsize; - ERTS_GET_REAL_BIN(list, real_bin, offset, bitoffs, bitsize); - bptr = binary_val(real_bin); - if (*bptr == HEADER_PROC_BIN && bitoffs == 0) { - size = binary_size(list); - vsize = 1; - } else - bptr = NULL; - } - - if (!bptr) { - if (io_list_vec_len(list, &vsize, &csize, &pvsize, &pcsize, &size)) - goto bad_value; + if (erts_ioq_iodata_vec_len(list, &vsize, &csize, &pvsize, &pcsize, + &size, ERL_SMALL_IO_BIN_LIMIT)) + goto bad_value; - /* To pack or not to pack (small binaries) ...? */ - if (vsize >= SMALL_WRITE_VEC) { - /* Do pack */ - vsize = pvsize + 1; - csize = pcsize; - blimit = ERL_SMALL_IO_BIN_LIMIT; - } - cbin = driver_alloc_binary(csize); + /* To pack or not to pack (small binaries) ...? */ + if (vsize >= SMALL_WRITE_VEC) { + /* Do pack */ + vsize = pvsize + 1; + csize = pcsize; + blimit = ERL_SMALL_IO_BIN_LIMIT; + } + if (csize) { + cbin = (ErtsIOQBinary *)driver_alloc_binary(csize); if (!cbin) erts_alloc_enomem(ERTS_ALC_T_DRV_BINARY, ERTS_SIZEOF_Binary(csize)); } - iov_offset = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErlIOVec)); binv_offset = iov_offset; binv_offset += ERTS_ALC_DATA_ALIGN_SIZE((vsize+1)*sizeof(SysIOVec)); alloc_size = binv_offset; - alloc_size += (vsize+1)*sizeof(ErlDrvBinary *); + alloc_size += (vsize+1)*sizeof(ErtsIOQBinary *); sigdp = erts_port_task_alloc_p2p_sig_data_extra(alloc_size, (void**)&ptr); - evp = (ErlIOVec *) ptr; - ivp = evp->iov = (SysIOVec *) (ptr + iov_offset); - bvp = evp->binv = (ErlDrvBinary **) (ptr + binv_offset); + evp = (ErtsIOVec *) ptr; + ivp = evp->driver.iov = (SysIOVec *) (ptr + iov_offset); + bvp = evp->common.binv = (ErtsIOQBinary **) (ptr + binv_offset); ivp[0].iov_base = NULL; ivp[0].iov_len = 0; bvp[0] = NULL; - if (bptr) { - ProcBin* pb = (ProcBin *) bptr; - - ivp[1].iov_base = pb->bytes+offset; - ivp[1].iov_len = size; - bvp[1] = Binary2ErlDrvBinary(pb->val); - - evp->vsize = 1; - } else { - - evp->vsize = io_list_to_vec(list, ivp+1, bvp+1, cbin, blimit); - if (evp->vsize < 0) { - if (evp != &ev) - erts_free(ERTS_ALC_T_DRV_CMD_DATA, evp); - driver_free_binary(cbin); - goto bad_value; - } + evp->driver.vsize = erts_ioq_iodata_to_vec(list, ivp+1, bvp+1, cbin, + blimit, 1); + if (evp->driver.vsize < 0) { + erts_free(ERTS_ALC_T_DRV_CMD_DATA, evp); + driver_free_binary(&cbin->driver); + goto bad_value; } #if 0 /* This assertion may say something useful, but it can be falsified during the emulator test suites. */ ASSERT(evp->vsize == vsize); #endif - evp->vsize++; - evp->size = size; /* total size */ + evp->driver.vsize++; + evp->driver.size = size; /* total size */ /* Need to increase refc on all binaries */ - for (i = 1; i < evp->vsize; i++) + for (i = 1; i < evp->driver.vsize; i++) if (bvp[i]) - driver_binary_inc_refc(bvp[i]); + driver_binary_inc_refc(&bvp[i]->driver); sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV; sigdp->u.outputv.from = from; - sigdp->u.outputv.evp = evp; - sigdp->u.outputv.cbinp = cbin; + sigdp->u.outputv.evp = &evp->driver; + sigdp->u.outputv.cbinp = &cbin->driver; port_sig_callback = port_sig_outputv; } else { ErlDrvSizeT ERTS_DECLARE_DUMMY(r); @@ -2102,26 +1736,18 @@ erts_port_output_async(Port *prt, Eterm from, Eterm list) sigdp->u.output.size = size; port_sig_callback = port_sig_output; } - sigdp->flags = 0; ns_pthp = NULL; task_flags = 0; - res = erts_schedule_proc2port_signal(NULL, - prt, - ERTS_INVALID_PID, - NULL, - sigdp, - task_flags, - ns_pthp, - port_sig_callback); + erts_schedule_proc2port_signal(NULL, + prt, + ERTS_INVALID_PID, + NULL, + sigdp, + task_flags, + ns_pthp, + port_sig_callback); - if (res != ERTS_PORT_OP_SCHEDULED) { - if (drv->outputv) - cleanup_scheduled_outputv(evp, cbin); - else - cleanup_scheduled_output(buf); - return 1; - } return 1; bad_value: @@ -2155,8 +1781,8 @@ erts_port_output(Process *c_p, erts_aint32_t sched_flags, busy_flgs, invalid_flags; int task_flags; ErtsProc2PortSigCallback port_sig_callback; - ErlDrvBinary *cbin = NULL; - ErlIOVec *evp = NULL; + ErtsIOQBinary *cbin = NULL; + ErtsIOVec *evp = NULL; char *buf = NULL; int force_immediate_call = (flags & ERTS_PORT_SIG_FLG_FORCE_IMM_CALL); int async_nosuspend; @@ -2202,11 +1828,11 @@ erts_port_output(Process *c_p, } #endif if (drv->outputv) { - ErlIOVec ev; + ErtsIOVec ev; SysIOVec iv[SMALL_WRITE_VEC]; - ErlDrvBinary* bv[SMALL_WRITE_VEC]; + ErtsIOQBinary* bv[SMALL_WRITE_VEC]; SysIOVec* ivp; - ErlDrvBinary** bvp; + ErtsIOQBinary** bvp; int vsize; Uint csize; Uint pvsize; @@ -2214,18 +1840,19 @@ erts_port_output(Process *c_p, Uint blimit; size_t iov_offset, binv_offset, alloc_size; - if (io_list_vec_len(list, &vsize, &csize, &pvsize, &pcsize, &size)) + if (erts_ioq_iodata_vec_len(list, &vsize, &csize, &pvsize, &pcsize, + &size, ERL_SMALL_IO_BIN_LIMIT)) goto bad_value; iov_offset = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErlIOVec)); binv_offset = iov_offset; binv_offset += ERTS_ALC_DATA_ALIGN_SIZE((vsize+1)*sizeof(SysIOVec)); alloc_size = binv_offset; - alloc_size += (vsize+1)*sizeof(ErlDrvBinary *); + alloc_size += (vsize+1)*sizeof(ErtsIOQBinary *); if (try_call && vsize < SMALL_WRITE_VEC) { - ivp = ev.iov = iv; - bvp = ev.binv = bv; + ivp = ev.common.iov = iv; + bvp = ev.common.binv = bv; evp = &ev; } else { @@ -2236,9 +1863,9 @@ erts_port_output(Process *c_p, sigdp = erts_port_task_alloc_p2p_sig_data_extra( alloc_size, (void**)&ptr); } - evp = (ErlIOVec *) ptr; - ivp = evp->iov = (SysIOVec *) (ptr + iov_offset); - bvp = evp->binv = (ErlDrvBinary **) (ptr + binv_offset); + evp = (ErtsIOVec *) ptr; + ivp = evp->driver.iov = (SysIOVec *) (ptr + iov_offset); + bvp = evp->common.binv = (ErtsIOQBinary **) (ptr + binv_offset); } /* To pack or not to pack (small binaries) ...? */ @@ -2254,23 +1881,26 @@ erts_port_output(Process *c_p, } /* Use vsize and csize from now on */ - cbin = driver_alloc_binary(csize); - if (!cbin) - erts_alloc_enomem(ERTS_ALC_T_DRV_BINARY, ERTS_SIZEOF_Binary(csize)); + if (csize) { + cbin = (ErtsIOQBinary *)driver_alloc_binary(csize); + if (!cbin) + erts_alloc_enomem(ERTS_ALC_T_DRV_BINARY, ERTS_SIZEOF_Binary(csize)); + } /* Element 0 is for driver usage to add header block */ ivp[0].iov_base = NULL; ivp[0].iov_len = 0; bvp[0] = NULL; - evp->vsize = io_list_to_vec(list, ivp+1, bvp+1, cbin, blimit); - if (evp->vsize < 0) { + evp->driver.vsize = erts_ioq_iodata_to_vec(list, ivp+1, bvp+1, + cbin, blimit, 1); + if (evp->driver.vsize < 0) { if (evp != &ev) { if (try_call) erts_free(ERTS_ALC_T_TMP, evp); else erts_port_task_free_p2p_sig_data(sigdp); } - driver_free_binary(cbin); + driver_free_binary(&cbin->driver); goto bad_value; } #if 0 @@ -2278,19 +1908,19 @@ erts_port_output(Process *c_p, be falsified during the emulator test suites. */ ASSERT(evp->vsize == vsize); #endif - evp->vsize++; - evp->size = size; /* total size */ + evp->driver.vsize++; + evp->driver.size = size; /* total size */ if (!try_call) { int i; /* Need to increase refc on all binaries */ - for (i = 1; i < evp->vsize; i++) - if (bvp[i]) - driver_binary_inc_refc(bvp[i]); + for (i = 1; i < evp->driver.vsize; i++) + if (bvp[i]) + driver_binary_inc_refc(&bvp[i]->driver); } else { int i; - ErlIOVec *new_evp; + ErtsIOVec *new_evp; ErtsTryImmDrvCallResult try_call_res; ErtsTryImmDrvCallState try_call_state = ERTS_INIT_TRY_IMM_DRV_CALL_STATE( @@ -2313,14 +1943,14 @@ erts_port_output(Process *c_p, from, prt, drv, - evp); + &evp->driver); if (force_immediate_call) finalize_force_imm_drv_call(&try_call_state); else finalize_imm_drv_call(&try_call_state); /* Fall through... */ case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: - driver_free_binary(cbin); + driver_free_binary(&cbin->driver); if (evp != &ev) { ASSERT(!sigdp); erts_free(ERTS_ALC_T_TMP, evp); @@ -2334,7 +1964,7 @@ erts_port_output(Process *c_p, sched_flags = try_call_state.sched_flags; if (async_nosuspend && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) { - driver_free_binary(cbin); + driver_free_binary(&cbin->driver); if (evp != &ev) { ASSERT(!sigdp); erts_free(ERTS_ALC_T_TMP, evp); @@ -2349,9 +1979,9 @@ erts_port_output(Process *c_p, } /* Need to increase refc on all binaries */ - for (i = 1; i < evp->vsize; i++) + for (i = 1; i < evp->driver.vsize; i++) if (bvp[i]) - driver_binary_inc_refc(bvp[i]); + driver_binary_inc_refc(&bvp[i]->driver); /* The port task and iovec is allocated in the same structure as an optimization. This @@ -2364,18 +1994,18 @@ erts_port_output(Process *c_p, if (evp != &ev) { /* Copy from TMP alloc to port task */ sys_memcpy((void *) new_evp, (void *) evp, alloc_size); - new_evp->iov = (SysIOVec *) (((char *) new_evp) - + iov_offset); - bvp = new_evp->binv = (ErlDrvBinary **) (((char *) new_evp) - + binv_offset); + new_evp->driver.iov = (SysIOVec *) (((char *) new_evp) + + iov_offset); + bvp = new_evp->common.binv = (ErtsIOQBinary **) (((char *) new_evp) + + binv_offset); #ifdef DEBUG - ASSERT(new_evp->vsize == evp->vsize); - ASSERT(new_evp->size == evp->size); - for (i = 0; i < evp->vsize; i++) { - ASSERT(new_evp->iov[i].iov_len == evp->iov[i].iov_len); - ASSERT(new_evp->iov[i].iov_base == evp->iov[i].iov_base); - ASSERT(new_evp->binv[i] == evp->binv[i]); + ASSERT(new_evp->driver.vsize == evp->driver.vsize); + ASSERT(new_evp->driver.size == evp->driver.size); + for (i = 0; i < evp->driver.vsize; i++) { + ASSERT(new_evp->driver.iov[i].iov_len == evp->driver.iov[i].iov_len); + ASSERT(new_evp->driver.iov[i].iov_base == evp->driver.iov[i].iov_base); + ASSERT(new_evp->driver.binv[i] == evp->driver.binv[i]); } #endif @@ -2384,24 +2014,24 @@ erts_port_output(Process *c_p, else { /* from stack allocated structure; offsets may differ */ sys_memcpy((void *) new_evp, (void *) evp, sizeof(ErlIOVec)); - new_evp->iov = (SysIOVec *) (((char *) new_evp) - + iov_offset); - sys_memcpy((void *) new_evp->iov, - (void *) evp->iov, - evp->vsize * sizeof(SysIOVec)); - new_evp->binv = (ErlDrvBinary **) (((char *) new_evp) - + binv_offset); - sys_memcpy((void *) new_evp->binv, - (void *) evp->binv, - evp->vsize * sizeof(ErlDrvBinary *)); + new_evp->driver.iov = (SysIOVec *) (((char *) new_evp) + + iov_offset); + sys_memcpy((void *) new_evp->driver.iov, + (void *) evp->driver.iov, + evp->driver.vsize * sizeof(SysIOVec)); + new_evp->common.binv = (ErtsIOQBinary **) (((char *) new_evp) + + binv_offset); + sys_memcpy((void *) new_evp->common.binv, + (void *) evp->common.binv, + evp->driver.vsize * sizeof(ErtsIOQBinary *)); #ifdef DEBUG - ASSERT(new_evp->vsize == evp->vsize); - ASSERT(new_evp->size == evp->size); - for (i = 0; i < evp->vsize; i++) { - ASSERT(new_evp->iov[i].iov_len == evp->iov[i].iov_len); - ASSERT(new_evp->iov[i].iov_base == evp->iov[i].iov_base); - ASSERT(new_evp->binv[i] == evp->binv[i]); + ASSERT(new_evp->driver.vsize == evp->driver.vsize); + ASSERT(new_evp->driver.size == evp->driver.size); + for (i = 0; i < evp->driver.vsize; i++) { + ASSERT(new_evp->driver.iov[i].iov_len == evp->driver.iov[i].iov_len); + ASSERT(new_evp->driver.iov[i].iov_base == evp->driver.iov[i].iov_base); + ASSERT(new_evp->driver.binv[i] == evp->driver.binv[i]); } #endif @@ -2412,8 +2042,8 @@ erts_port_output(Process *c_p, sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV; sigdp->u.outputv.from = from; - sigdp->u.outputv.evp = evp; - sigdp->u.outputv.cbinp = cbin; + sigdp->u.outputv.evp = &evp->driver; + sigdp->u.outputv.cbinp = &cbin->driver; port_sig_callback = port_sig_outputv; } else { @@ -2554,10 +2184,6 @@ erts_port_output(Process *c_p, port_sig_callback); if (res != ERTS_PORT_OP_SCHEDULED) { - if (drv->outputv) - cleanup_scheduled_outputv(evp, cbin); - else - cleanup_scheduled_output(buf); return res; } @@ -2736,21 +2362,14 @@ erts_port_exit(Process *c_p, &bp->off_heap); } - res = erts_schedule_proc2port_signal(c_p, - prt, - c_p ? c_p->common.id : from, - refp, - sigdp, - 0, - NULL, - port_sig_exit); - - if (res == ERTS_PORT_OP_DROPPED) { - if (bp) - free_message_buffer(bp); - } - - return res; + return erts_schedule_proc2port_signal(c_p, + prt, + c_p ? c_p->common.id : from, + refp, + sigdp, + 0, + NULL, + port_sig_exit); } static ErtsPortOpResult @@ -3101,7 +2720,7 @@ port_monitor(Port *prt, erts_aint32_t state, Eterm origin, ASSERT(is_pid(origin)); ASSERT(is_atom(name) || is_port(name) || name == NIL); - ASSERT(is_internal_ref(ref)); + ASSERT(is_internal_ordinary_ref(ref)); if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) { ErtsProcLocks p_locks = ERTS_PROC_LOCK_LINK; @@ -3126,7 +2745,7 @@ static int port_sig_monitor(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *sigdp) { - Eterm hp[REF_THING_SIZE]; + Eterm hp[ERTS_REF_THING_SIZE]; Eterm ref = make_internal_ref(&hp); write_ref_thing(hp, sigdp->ref[0], sigdp->ref[1], sigdp->ref[2]); @@ -3247,7 +2866,7 @@ static int port_sig_demonitor(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *sigdp) { - Eterm hp[REF_THING_SIZE]; + Eterm hp[ERTS_REF_THING_SIZE]; Eterm ref = make_internal_ref(&hp); write_ref_thing(hp, sigdp->u.demonitor.ref[0], sigdp->u.demonitor.ref[1], @@ -3304,10 +2923,10 @@ ErtsPortOpResult erts_port_demonitor(Process *origin, ErtsDemonitorMode mode, sigdp->u.demonitor.origin = origin->common.id; sigdp->u.demonitor.name = target->common.id; { - RefThing *reft = ref_thing_ptr(ref); + Uint32 *nums = internal_ref_numbers(ref); /* Start from 1 skip ref arity */ sys_memcpy(sigdp->u.demonitor.ref, - internal_thing_ref_numbers(reft), + nums, sizeof(sigdp->u.demonitor.ref)); } @@ -3419,9 +3038,8 @@ void erts_init_io(int port_tab_size, else if (port_tab_size < ERTS_MIN_PORTS) port_tab_size = ERTS_MIN_PORTS; - erts_smp_rwmtx_init_opt(&erts_driver_list_lock, - &drv_list_rwmtx_opts, - "driver_list"); + erts_smp_rwmtx_init_opt(&erts_driver_list_lock, &drv_list_rwmtx_opts, "driver_list", NIL, + ERTS_LOCK_FLAGS_PROPERTY_STATIC | ERTS_LOCK_FLAGS_CATEGORY_IO); driver_list = NULL; erts_smp_tsd_key_create(&driver_list_lock_status_key, "erts_driver_list_lock_status_key"); @@ -3433,7 +3051,7 @@ void erts_init_io(int port_tab_size, NULL, (ErtsPTabElementCommon *) &erts_invalid_port.common, port_tab_size, - common_element_size, /* Doesn't need to be excact */ + common_element_size, /* Doesn't need to be exact */ "port_table", legacy_port_tab, 1); @@ -3458,67 +3076,94 @@ void erts_init_io(int port_tab_size, } #if defined(ERTS_ENABLE_LOCK_COUNT) && defined(ERTS_SMP) -static ERTS_INLINE void lcnt_enable_drv_lock_count(erts_driver_t *dp, int enable) +static void lcnt_enable_driver_lock_count(erts_driver_t *dp, int enable) { if (dp->lock) { - if (enable) - erts_lcnt_init_lock_x(&dp->lock->lcnt, - "driver_lock", - ERTS_LCNT_LT_MUTEX, - erts_atom_put((byte*)dp->name, - sys_strlen(dp->name), - ERTS_ATOM_ENC_LATIN1, - 1)); - - else - erts_lcnt_destroy_lock(&dp->lock->lcnt); + if (enable) { + Eterm name_as_atom = erts_atom_put((byte*)dp->name, sys_strlen(dp->name), + ERTS_ATOM_ENC_LATIN1, 1); + erts_lcnt_install_new_lock_info(&dp->lock->lcnt, "driver_lock", name_as_atom, + ERTS_LOCK_TYPE_MUTEX | ERTS_LOCK_FLAGS_CATEGORY_IO); + } else { + erts_lcnt_uninstall(&dp->lock->lcnt); + } } } -static ERTS_INLINE void lcnt_enable_port_lock_count(Port *prt, int enable) +static void lcnt_enable_port_lock_count(Port *prt, int enable) { erts_aint32_t state = erts_atomic32_read_nob(&prt->state); - if (!enable) { - erts_lcnt_destroy_lock(&prt->sched.mtx.lcnt); - if (state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) - erts_lcnt_destroy_lock(&prt->lock->lcnt); + + if(enable) { + ErlDrvPDL pdl = prt->port_data_lock; + + erts_lcnt_install_new_lock_info(&prt->sched.mtx.lcnt, "port_sched_lock", + prt->common.id, ERTS_LOCK_TYPE_MUTEX | ERTS_LOCK_FLAGS_CATEGORY_IO); + + if(pdl) { + erts_lcnt_install_new_lock_info(&pdl->mtx.lcnt, "port_data_lock", + prt->common.id, ERTS_LOCK_TYPE_MUTEX | ERTS_LOCK_FLAGS_CATEGORY_IO); + } + + if(state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) { + erts_lcnt_install_new_lock_info(&prt->lock->lcnt, "port_lock", + prt->common.id, ERTS_LOCK_TYPE_MUTEX | ERTS_LOCK_FLAGS_CATEGORY_IO); + } + } else { + erts_lcnt_uninstall(&prt->sched.mtx.lcnt); + + if(prt->port_data_lock) { + erts_lcnt_uninstall(&prt->port_data_lock->mtx.lcnt); + } + + if(state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) { + erts_lcnt_uninstall(&prt->lock->lcnt); + } } - else { - erts_lcnt_init_lock_x(&prt->sched.mtx.lcnt, - "port_sched_lock", - ERTS_LCNT_LT_MUTEX, - prt->common.id); - if (state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) - erts_lcnt_init_lock_x(&prt->lock->lcnt, - "port_lock", - ERTS_LCNT_LT_MUTEX, - prt->common.id); +} + +void erts_lcnt_update_driver_locks(int enable) { + erts_driver_t *driver; + + lcnt_enable_driver_lock_count(&vanilla_driver, enable); + lcnt_enable_driver_lock_count(&spawn_driver, enable); +#ifndef __WIN32__ + lcnt_enable_driver_lock_count(&forker_driver, enable); +#endif + lcnt_enable_driver_lock_count(&fd_driver, enable); + + erts_rwmtx_rlock(&erts_driver_list_lock); + + for (driver = driver_list; driver; driver = driver->next) { + lcnt_enable_driver_lock_count(driver, enable); } + + erts_rwmtx_runlock(&erts_driver_list_lock); } -void erts_lcnt_enable_io_lock_count(int enable) { - erts_driver_t *dp; - int ix, max = erts_ptab_max(&erts_port); - Port *prt; +void erts_lcnt_update_port_locks(int enable) { + int i, max; - for (ix = 0; ix < max; ix++) { - if ((prt = erts_pix2port(ix)) != NULL) { - lcnt_enable_port_lock_count(prt, enable); + max = erts_ptab_max(&erts_port); + + for(i = 0; i < max; i++) { + int delay_handle; + Port *port; + + delay_handle = erts_thr_progress_unmanaged_delay(); + port = erts_pix2port(i); + + if(port != NULL) { + lcnt_enable_port_lock_count(port, enable); } - } /* for all ports */ - lcnt_enable_drv_lock_count(&vanilla_driver, enable); - lcnt_enable_drv_lock_count(&spawn_driver, enable); -#ifndef __WIN32__ - lcnt_enable_drv_lock_count(&forker_driver, enable); -#endif - lcnt_enable_drv_lock_count(&fd_driver, enable); - /* enable lock counting in all drivers */ - for (dp = driver_list; dp; dp = dp->next) { - lcnt_enable_drv_lock_count(dp, enable); + if(delay_handle != ERTS_THR_PRGR_DHANDLE_MANAGED) { + erts_thr_progress_unmanaged_continue(delay_handle); + } } -} /* enable/disable lock counting of ports */ +} + #endif /* defined(ERTS_ENABLE_LOCK_COUNT) && defined(ERTS_SMP) */ /* * Buffering of data when using line oriented I/O on ports @@ -3701,7 +3346,7 @@ deliver_result(Port *prt, Eterm sender, Eterm pid, Eterm res) ERTS_SMP_CHK_NO_PROC_LOCKS; ASSERT(!prt || prt->common.id == sender); -#ifdef ERTS_SMP +#if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK) ASSERT(!prt || erts_lc_is_port_locked(prt)); #endif @@ -3793,7 +3438,6 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to, Binary* bptr; bptr = erts_bin_nrml_alloc(len); - erts_refc_init(&bptr->refc, 1); sys_memcpy(bptr->orig_bytes, buf, len); pb = (ProcBin *) hp; @@ -4194,8 +3838,8 @@ static void sweep_one_monitor(ErtsMonitor *mon, void *vpsc) ErtsMonitor *rmon; Process *rp; - ASSERT(is_internal_pid(mon->pid)); - rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK); + ASSERT(is_internal_pid(mon->u.pid)); + rp = erts_pid2proc(NULL, 0, mon->u.pid, ERTS_PROC_LOCK_LINK); if (!rp) { goto done; } @@ -4296,7 +3940,7 @@ port_fire_one_monitor(ErtsMonitor *mon, void *ctx0) Process *origin; ErtsProcLocks origin_locks; - if (mon->type != MON_TARGET || ! is_pid(mon->pid)) { + if (mon->type != MON_TARGET || ! is_pid(mon->u.pid)) { return; } /* @@ -4305,7 +3949,7 @@ port_fire_one_monitor(ErtsMonitor *mon, void *ctx0) */ origin_locks = ERTS_PROC_LOCKS_MSG_SEND | ERTS_PROC_LOCK_LINK; - origin = erts_pid2proc(NULL, 0, mon->pid, origin_locks); + origin = erts_pid2proc(NULL, 0, mon->u.pid, origin_locks); if (origin) { DeclareTmpHeapNoproc(lhp,3); SweepContext *ctx = (SweepContext *)ctx0; @@ -4560,8 +4204,7 @@ static void cleanup_scheduled_control(Binary *binp, char *bufp) { if (binp) { - if (erts_refc_dectest(&binp->refc, 0) == 0) - erts_bin_free(binp); + erts_bin_release(binp); } else { if (bufp) @@ -4905,7 +4548,7 @@ erts_port_control(Process* c_p, ASSERT(bufp <= bufp + size); ASSERT(binp->orig_bytes <= bufp && bufp + size <= binp->orig_bytes + binp->orig_size); - erts_refc_inc(&binp->refc, 1); + erts_refc_inc(&binp->intern.refc, 1); } } @@ -4932,10 +4575,9 @@ erts_port_control(Process* c_p, 0, NULL, port_sig_control); - if (res != ERTS_PORT_OP_SCHEDULED) { - cleanup_scheduled_control(binp, bufp); + if (res != ERTS_PORT_OP_SCHEDULED) return ERTS_PORT_OP_BADARG; - } + return res; } @@ -5225,10 +4867,9 @@ erts_port_call(Process* c_p, 0, NULL, port_sig_call); - if (res != ERTS_PORT_OP_SCHEDULED) { - cleanup_scheduled_call(bufp); + if (res != ERTS_PORT_OP_SCHEDULED) return ERTS_PORT_OP_BADARG; - } + return res; } @@ -5429,7 +5070,7 @@ reply_io_bytes(void *vreq) rp_locks = ERTS_PROC_LOCK_MAIN; } - hsz = 5 /* 4-tuple */ + REF_THING_SIZE; + hsz = 5 /* 4-tuple */ + ERTS_REF_THING_SIZE; erts_bld_uint64(NULL, &hsz, in); erts_bld_uint64(NULL, &hsz, out); @@ -5438,7 +5079,7 @@ reply_io_bytes(void *vreq) ref = make_internal_ref(hp); write_ref_thing(hp, req->refn[0], req->refn[1], req->refn[2]); - hp += REF_THING_SIZE; + hp += ERTS_REF_THING_SIZE; ein = erts_bld_uint64(&hp, NULL, in); eout = erts_bld_uint64(&hp, NULL, out); @@ -5467,7 +5108,7 @@ erts_request_io_bytes(Process *c_p) ErtsIOBytesReq *req = erts_alloc(ERTS_ALC_T_IOB_REQ, sizeof(ErtsIOBytesReq)); - hp = HAlloc(c_p, REF_THING_SIZE); + hp = HAlloc(c_p, ERTS_REF_THING_SIZE); ref = erts_sched_make_ref_in_buffer(esdp, hp); refn = internal_ref_numbers(ref); @@ -5501,7 +5142,7 @@ typedef struct { static void prt_one_monitor(ErtsMonitor *mon, void *vprtd) { prt_one_lnk_data *prtd = (prt_one_lnk_data *) vprtd; - erts_print(prtd->to, prtd->arg, "(%T,%T)", mon->pid,mon->ref); + erts_print(prtd->to, prtd->arg, "(%T,%T)", mon->u.pid, mon->ref); } static void prt_one_lnk(ErtsLink *lnk, void *vprtd) @@ -5510,6 +5151,93 @@ static void prt_one_lnk(ErtsLink *lnk, void *vprtd) erts_print(prtd->to, prtd->arg, "%T", lnk->pid); } +static void dump_port_state(fmtfn_t to, void *arg, erts_aint32_t state) +{ + erts_aint32_t rest; + int unknown = 0; + char delim = ' '; + + erts_print(to, arg, "State:"); + + rest = state; + while (rest) { + erts_aint32_t chk = (rest ^ (rest-1)) & rest; /* lowest set bit */ + char* s; + + rest &= ~chk; + switch (chk) { + case ERTS_PORT_SFLG_CONNECTED: s = "CONNECTED"; break; + case ERTS_PORT_SFLG_EXITING: s = "EXITING"; break; + case ERTS_PORT_SFLG_DISTRIBUTION: s = "DISTR"; break; + case ERTS_PORT_SFLG_BINARY_IO: s = "BINARY_IO"; break; + case ERTS_PORT_SFLG_SOFT_EOF: s = "SOFT_EOF"; break; + case ERTS_PORT_SFLG_CLOSING: s = "CLOSING"; break; + case ERTS_PORT_SFLG_SEND_CLOSED: s = "SEND_CLOSED"; break; + case ERTS_PORT_SFLG_LINEBUF_IO: s = "LINEBUF_IO"; break; + case ERTS_PORT_SFLG_FREE: s = "FREE"; break; + case ERTS_PORT_SFLG_INITIALIZING: s = "INITIALIZING"; break; + case ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK: s = "PORT_LOCK"; break; + case ERTS_PORT_SFLG_INVALID: s = "INVALID"; break; + case ERTS_PORT_SFLG_HALT: s = "HALT"; break; +#ifdef DEBUG + case ERTS_PORT_SFLG_PORT_DEBUG: s = "DEBUG"; break; +#endif + default: + unknown = 1; + continue; + } + erts_print(to, arg, "%c%s", delim, s); + delim = '|'; + } + if (unknown || !state) + erts_print(to, arg, "%c0x%x\n", delim, state); + else + erts_print(to, arg, "\n"); +} + +static void dump_port_task_flags(fmtfn_t to, void *arg, Port* p) +{ + erts_aint32_t flags = erts_smp_atomic32_read_nob(&p->sched.flags); + erts_aint32_t unknown = 0; + char delim = ' '; + + if (!flags) + return; + + erts_print(to, arg, "Task Flags:"); + + while (flags) { + erts_aint32_t chk = (flags ^ (flags-1)) & flags; /* lowest set bit */ + char* s; + + flags &= ~chk; + switch (chk) { + case ERTS_PTS_FLG_IN_RUNQ: s = "IN_RUNQ"; break; + case ERTS_PTS_FLG_EXEC: s = "EXEC"; break; + case ERTS_PTS_FLG_HAVE_TASKS: s = "HAVE_TASKS"; break; + case ERTS_PTS_FLG_EXIT: s = "EXIT"; break; + case ERTS_PTS_FLG_BUSY_PORT: s = "BUSY_PORT"; break; + case ERTS_PTS_FLG_BUSY_PORT_Q: s = "BUSY_Q"; break; + case ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q: s = "CHK_UNSET_BUSY_Q"; break; + case ERTS_PTS_FLG_HAVE_BUSY_TASKS: s = "BUSY_TASKS"; break; + case ERTS_PTS_FLG_HAVE_NS_TASKS: s = "NS_TASKS"; break; + case ERTS_PTS_FLG_PARALLELISM: s = "PARALLELISM"; break; + case ERTS_PTS_FLG_FORCE_SCHED: s = "FORCE_SCHED"; break; + case ERTS_PTS_FLG_EXITING: s = "EXITING"; break; + case ERTS_PTS_FLG_EXEC_IMM: s = "EXEC_IMM"; break; + default: + unknown |= chk; + continue; + } + erts_print(to, arg, "%c%s", delim, s); + delim = '|'; + } + if (unknown) + erts_print(to, arg, "%cUNKNOWN(0x%x)\n", delim, unknown); + else + erts_print(to, arg, "\n"); +} + void print_port_info(Port *p, fmtfn_t to, void *arg) { @@ -5519,6 +5247,8 @@ print_port_info(Port *p, fmtfn_t to, void *arg) return; erts_print(to, arg, "=port:%T\n", p->common.id); + dump_port_state(to, arg, state); + dump_port_task_flags(to, arg, p); erts_print(to, arg, "Slot: %d\n", internal_port_index(p->common.id)); if (state & ERTS_PORT_SFLG_CONNECTED) { erts_print(to, arg, "Connected: %T", ERTS_PORT_GET_CONNECTED(p)); @@ -5541,6 +5271,10 @@ print_port_info(Port *p, fmtfn_t to, void *arg) erts_doforall_monitors(ERTS_P_MONITORS(p), &prt_one_monitor, &prtd); erts_print(to, arg, "\n"); } + if (p->suspended) { + erts_print(to, arg, "Suspended: "); + erts_proclist_dump(to, arg, p->suspended); + } if (p->common.u.alive.reg != NULL) erts_print(to, arg, "Registered as: %T\n", p->common.u.alive.reg->name); @@ -5558,6 +5292,14 @@ print_port_info(Port *p, fmtfn_t to, void *arg) } else { erts_print(to, arg, "Port controls linked-in driver: %s\n",p->name); } + erts_print(to, arg, "Input: %beu\n", p->bytes_in); + erts_print(to, arg, "Output: %beu\n", p->bytes_out); + erts_print(to, arg, "Queue: %beu\n", erts_ioq_size(&p->ioq)); + { + Eterm port_data = erts_port_data_read(p); + if (port_data != am_undefined) + erts_print(to, arg, "Port Data: %T\n", port_data); + } } void @@ -6402,7 +6144,6 @@ driver_deliver_term(Port *prt, Eterm to, ErlDrvTermData* data, int len) ProcBin* pbp; Binary* bp = erts_bin_nrml_alloc(size); ASSERT(bufp); - erts_refc_init(&bp->refc, 1); sys_memcpy((void *) bp->orig_bytes, (void *) bufp, size); pbp = (ProcBin *) erts_produce_heap(&factory, PROC_BIN_SIZE, HEAP_EXTRA); @@ -6912,21 +6653,21 @@ ErlDrvSInt driver_binary_get_refc(ErlDrvBinary *dbp) { Binary* bp = ErlDrvBinary2Binary(dbp); - return (ErlDrvSInt) erts_refc_read(&bp->refc, 1); + return (ErlDrvSInt) erts_refc_read(&bp->intern.refc, 1); } ErlDrvSInt driver_binary_inc_refc(ErlDrvBinary *dbp) { Binary* bp = ErlDrvBinary2Binary(dbp); - return (ErlDrvSInt) erts_refc_inctest(&bp->refc, 2); + return (ErlDrvSInt) erts_refc_inctest(&bp->intern.refc, 2); } ErlDrvSInt driver_binary_dec_refc(ErlDrvBinary *dbp) { Binary* bp = ErlDrvBinary2Binary(dbp); - return (ErlDrvSInt) erts_refc_dectest(&bp->refc, 1); + return (ErlDrvSInt) erts_refc_dectest(&bp->intern.refc, 1); } @@ -6942,7 +6683,6 @@ driver_alloc_binary(ErlDrvSizeT size) bin = erts_bin_drv_alloc_fnf((Uint) size); if (!bin) return NULL; /* The driver write must take action */ - erts_refc_init(&bin->refc, 1); return Binary2ErlDrvBinary(bin); } @@ -6972,8 +6712,7 @@ void driver_free_binary(ErlDrvBinary* dbin) return; bin = ErlDrvBinary2Binary(dbin); - if (erts_refc_dectest(&bin->refc, 0) == 0) - erts_bin_free(bin); + erts_bin_release(bin); } @@ -7098,7 +6837,7 @@ driver_pdl_create(ErlDrvPort dp) return NULL; pdl = erts_alloc(ERTS_ALC_T_PORT_DATA_LOCK, sizeof(struct erl_drv_port_data_lock)); - erts_mtx_init_x(&pdl->mtx, "port_data_lock", pp->common.id, 1); + erts_mtx_init(&pdl->mtx, "port_data_lock", pp->common.id, ERTS_LOCK_FLAGS_CATEGORY_IO); pdl_init_refc(pdl); erts_port_inc_refc(pp); pdl->prt = pp; @@ -7162,307 +6901,51 @@ driver_pdl_dec_refc(ErlDrvPDL pdl) return refc; } -/* expand queue to hold n elements in tail or head */ -static int expandq(ErlIOQueue* 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; - ErlDrvBinary** 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(ERTS_ALC_T_IOQ, nvsz * sizeof(SysIOVec)); - if (!niov) - return -1; - nbinv = erts_alloc_fnf(ERTS_ALC_T_IOQ, nvsz * sizeof(ErlDrvBinary**)); - if (!nbinv) { - erts_free(ERTS_ALC_T_IOQ, (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(ERTS_ALC_T_IOQ, (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(ErlDrvBinary*)); - if (q->b_start != q->b_small) - erts_free(ERTS_ALC_T_IOQ, (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(ERTS_ALC_T_IOQ, (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(ErlDrvBinary*)); - if (q->b_start != q->b_small) - erts_free(ERTS_ALC_T_IOQ, (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(ErlDrvBinary*)); - 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(ErlDrvBinary*)); - q->b_tail = q->b_end; - q->b_head = q->b_tail-q_sz; - } - - return 0; -} - - - /* Put elements from vec at q tail */ int driver_enqv(ErlDrvPort ix, ErlIOVec* vec, ErlDrvSizeT skip) { - int n; - size_t len; - ErlDrvSizeT size; - SysIOVec* iov; - ErlDrvBinary** binv; - ErlDrvBinary* b; - ErlIOQueue* q = drvport2ioq(ix); - - if (q == NULL) - return -1; - - ASSERT(vec->size >= skip); /* debug only */ - if (vec->size <= skip) - return 0; - size = vec->size - skip; - - iov = vec->iov; - binv = vec->binv; - n = vec->vsize; - - /* we use do here to strip iov_len=0 from beginning */ - do { - len = iov->iov_len; - if (len <= skip) { - skip -= len; - iov++; - binv++; - n--; - } - else { - iov->iov_base = ((char *)(iov->iov_base)) + skip; - iov->iov_len -= skip; - skip = 0; - } - } while(skip > 0); - - if (q->v_tail + n >= q->v_end) - expandq(q, n, 1); - - /* Queue and reference all binaries (remove zero length items) */ - while(n--) { - if ((len = iov->iov_len) > 0) { - if ((b = *binv) == NULL) { /* speical case create binary ! */ - b = driver_alloc_binary(len); - sys_memcpy(b->orig_bytes, iov->iov_base, len); - *q->b_tail++ = b; - q->v_tail->iov_len = len; - q->v_tail->iov_base = b->orig_bytes; - q->v_tail++; - } - else { - driver_binary_inc_refc(b); - *q->b_tail++ = b; - *q->v_tail++ = *iov; - } - } - iov++; - binv++; - } - q->size += size; /* update total size in queue */ - return 0; + ASSERT(vec->size >= skip); + return erts_ioq_enqv(drvport2ioq(ix), (ErtsIOVec*)vec, skip); } /* Put elements from vec at q head */ int driver_pushqv(ErlDrvPort ix, ErlIOVec* vec, ErlDrvSizeT skip) { - int n; - size_t len; - ErlDrvSizeT size; - SysIOVec* iov; - ErlDrvBinary** binv; - ErlDrvBinary* b; - ErlIOQueue* q = drvport2ioq(ix); - - if (q == NULL) - return -1; - - if (vec->size <= skip) - return 0; - size = vec->size - skip; - - iov = vec->iov; - binv = vec->binv; - n = vec->vsize; - - /* we use do here to strip iov_len=0 from beginning */ - do { - len = iov->iov_len; - if (len <= skip) { - skip -= len; - iov++; - binv++; - n--; - } - else { - iov->iov_base = ((char *)(iov->iov_base)) + skip; - iov->iov_len -= skip; - skip = 0; - } - } while(skip > 0); - - if (q->v_head - n < q->v_start) - expandq(q, n, 0); - - /* 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) { /* speical case create binary ! */ - b = driver_alloc_binary(len); - sys_memcpy(b->orig_bytes, iov->iov_base, len); - *--q->b_head = b; - q->v_head--; - q->v_head->iov_len = len; - q->v_head->iov_base = b->orig_bytes; - } - else { - driver_binary_inc_refc(b); - *--q->b_head = b; - *--q->v_head = *iov; - } - } - iov--; - binv--; - } - q->size += size; /* update total size in queue */ - return 0; + ASSERT(vec->size >= skip); + return erts_ioq_pushqv(drvport2ioq(ix), (ErtsIOVec*)vec, skip); } - /* ** Remove size bytes from queue head ** Return number of bytes that remain in queue */ ErlDrvSizeT driver_deq(ErlDrvPort ix, ErlDrvSizeT size) { - ErlIOQueue* q = drvport2ioq(ix); - ErlDrvSizeT 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; - driver_free_binary(*q->b_head); - *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 q->size; + ErlPortIOQueue *q = drvport2ioq(ix); + if (erts_ioq_deq(q, size) == -1) + return -1; + return erts_ioq_size(q); } -ErlDrvSizeT driver_peekqv(ErlDrvPort ix, ErlIOVec *ev) { - ErlIOQueue *q = drvport2ioq(ix); - ASSERT(ev); - - if (! q) { - return (ErlDrvSizeT) -1; - } else { - if ((ev->vsize = q->v_tail - q->v_head) == 0) { - ev->size = 0; - ev->iov = NULL; - ev->binv = NULL; - } else { - ev->size = q->size; - ev->iov = q->v_head; - ev->binv = q->b_head; - } - return q->size; - } +ErlDrvSizeT driver_peekqv(ErlDrvPort ix, ErlIOVec *ev) +{ + return erts_ioq_peekqv(drvport2ioq(ix), (ErtsIOVec*)ev); } SysIOVec* driver_peekq(ErlDrvPort ix, int* vlenp) /* length of io-vector */ { - ErlIOQueue* q = drvport2ioq(ix); - - if (q == NULL) { - *vlenp = -1; - return NULL; - } - if ((*vlenp = (q->v_tail - q->v_head)) == 0) - return NULL; - return q->v_head; + return erts_ioq_peekq(drvport2ioq(ix), vlenp); } ErlDrvSizeT driver_sizeq(ErlDrvPort ix) { - ErlIOQueue* q = drvport2ioq(ix); + ErlPortIOQueue *q = drvport2ioq(ix); if (q == NULL) - return (size_t) -1; - return q->size; + return (ErlDrvSizeT) -1; + return erts_ioq_size(q); } @@ -7623,22 +7106,29 @@ erl_drv_convert_time_unit(ErlDrvTime val, (int) to); } -static void ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon) +void erts_ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon) { - RefThing *refp; - ASSERT(is_internal_ref(ref)); - ERTS_CT_ASSERT(sizeof(RefThing) <= sizeof(ErlDrvMonitor)); - refp = ref_thing_ptr(ref); - memset(mon,0,sizeof(ErlDrvMonitor)); - memcpy(mon,refp,sizeof(RefThing)); + ERTS_CT_ASSERT(ERTS_REF_THING_SIZE*sizeof(Uint) <= sizeof(ErlDrvMonitor)); + ASSERT(is_internal_ordinary_ref(ref)); + sys_memcpy((void *) mon, (void *) internal_ref_val(ref), + ERTS_REF_THING_SIZE*sizeof(Uint)); } +Eterm erts_driver_monitor_to_ref(Eterm *hp, const ErlDrvMonitor *mon) +{ + Eterm ref; + ERTS_CT_ASSERT(ERTS_REF_THING_SIZE*sizeof(Uint) <= sizeof(ErlDrvMonitor)); + sys_memcpy((void *) hp, (void *) mon, ERTS_REF_THING_SIZE*sizeof(Uint)); + ref = make_internal_ref(hp); + ASSERT(is_internal_ordinary_ref(ref)); + return ref; +} static int do_driver_monitor_process(Port *prt, - Eterm *buf, ErlDrvTermData process, ErlDrvMonitor *monitor) { + Eterm buf[ERTS_REF_THING_SIZE]; Process *rp; Eterm ref; @@ -7657,7 +7147,7 @@ static int do_driver_monitor_process(Port *prt, erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, ref, prt->common.id, NIL); erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - ref_to_driver_monitor(ref,monitor); + erts_ref_to_driver_monitor(ref,monitor); return 0; } @@ -7681,32 +7171,27 @@ int driver_monitor_process(ErlDrvPort drvport, /* Now (in SMP) we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); - { - DeclareTmpHeapNoproc(buf,REF_THING_SIZE); - UseTmpHeapNoproc(REF_THING_SIZE); - ret = do_driver_monitor_process(prt,buf,process,monitor); - UnUseTmpHeapNoproc(REF_THING_SIZE); - } + ret = do_driver_monitor_process(prt,process,monitor); DRV_MONITOR_UNLOCK_PDL(prt); return ret; } -static int do_driver_demonitor_process(Port *prt, Eterm *buf, - const ErlDrvMonitor *monitor) +static int do_driver_demonitor_process(Port *prt, const ErlDrvMonitor *monitor) { + Eterm heap[ERTS_REF_THING_SIZE]; Process *rp; Eterm ref; ErtsMonitor *mon; Eterm to; - memcpy(buf,monitor,sizeof(Eterm)*REF_THING_SIZE); - ref = make_internal_ref(buf); + ref = erts_driver_monitor_to_ref(heap, monitor); + mon = erts_lookup_monitor(ERTS_P_MONITORS(prt), ref); if (mon == NULL) { return 1; } ASSERT(mon->type == MON_ORIGIN); - to = mon->pid; + to = mon->u.pid; ASSERT(is_internal_pid(to)); rp = erts_pid2proc_opt(NULL, 0, @@ -7744,31 +7229,26 @@ int driver_demonitor_process(ErlDrvPort drvport, /* Now we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); - { - DeclareTmpHeapNoproc(buf,REF_THING_SIZE); - UseTmpHeapNoproc(REF_THING_SIZE); - ret = do_driver_demonitor_process(prt,buf,monitor); - UnUseTmpHeapNoproc(REF_THING_SIZE); - } + ret = do_driver_demonitor_process(prt,monitor); DRV_MONITOR_UNLOCK_PDL(prt); return ret; } -static ErlDrvTermData do_driver_get_monitored_process(Port *prt, Eterm *buf, - const ErlDrvMonitor *monitor) +static ErlDrvTermData do_driver_get_monitored_process(Port *prt,const ErlDrvMonitor *monitor) { Eterm ref; ErtsMonitor *mon; Eterm to; + Eterm heap[ERTS_REF_THING_SIZE]; + + ref = erts_driver_monitor_to_ref(heap, monitor); - memcpy(buf,monitor,sizeof(Eterm)*REF_THING_SIZE); - ref = make_internal_ref(buf); mon = erts_lookup_monitor(ERTS_P_MONITORS(prt), ref); if (mon == NULL) { return driver_term_nil; } ASSERT(mon->type == MON_ORIGIN); - to = mon->pid; + to = mon->u.pid; ASSERT(is_internal_pid(to)); return (ErlDrvTermData) to; } @@ -7790,21 +7270,16 @@ ErlDrvTermData driver_get_monitored_process(ErlDrvPort drvport, /* Now we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); - { - DeclareTmpHeapNoproc(buf,REF_THING_SIZE); - UseTmpHeapNoproc(REF_THING_SIZE); - ret = do_driver_get_monitored_process(prt,buf,monitor); - UnUseTmpHeapNoproc(REF_THING_SIZE); - } + ret = do_driver_get_monitored_process(prt,monitor); DRV_MONITOR_UNLOCK_PDL(prt); return ret; } - int driver_compare_monitors(const ErlDrvMonitor *monitor1, const ErlDrvMonitor *monitor2) { - return memcmp(monitor1,monitor2,sizeof(ErlDrvMonitor)); + return sys_memcmp((void *) monitor1, (void *) monitor2, + ERTS_REF_THING_SIZE*sizeof(Eterm)); } void erts_fire_port_monitor(Port *prt, Eterm ref) @@ -7824,7 +7299,7 @@ void erts_fire_port_monitor(Port *prt, Eterm ref) } callback = prt->drv_ptr->process_exit; ASSERT(callback != NULL); - ref_to_driver_monitor(ref,&drv_monitor); + erts_ref_to_driver_monitor(ref,&drv_monitor); ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); DRV_MONITOR_UNLOCK_PDL(prt); #ifdef USE_VM_PROBES @@ -8273,23 +7748,16 @@ init_driver(erts_driver_t *drv, ErlDrvEntry *de, DE_Handle *handle) drv->flags = de->driver_flags; drv->handle = handle; #ifdef ERTS_SMP - if (drv->flags & ERL_DRV_FLAG_USE_PORT_LOCKING) - drv->lock = NULL; - else { - drv->lock = erts_alloc(ERTS_ALC_T_DRIVER_LOCK, - sizeof(erts_mtx_t)); - erts_mtx_init_x(drv->lock, - "driver_lock", -#if defined(ERTS_ENABLE_LOCK_CHECK) || defined(ERTS_ENABLE_LOCK_COUNT) - erts_atom_put((byte *) drv->name, - sys_strlen(drv->name), - ERTS_ATOM_ENC_LATIN1, - 1), -#else - NIL, -#endif - 1 - ); + if (drv->flags & ERL_DRV_FLAG_USE_PORT_LOCKING) { + drv->lock = NULL; + } else { + Eterm driver_id = erts_atom_put((byte *) drv->name, + sys_strlen(drv->name), + ERTS_ATOM_ENC_LATIN1, 1); + + drv->lock = erts_alloc(ERTS_ALC_T_DRIVER_LOCK, sizeof(erts_mtx_t)); + + erts_mtx_init(drv->lock, "driver_lock", driver_id, ERTS_LOCK_FLAGS_CATEGORY_IO); } #endif drv->entry = de; |