aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator')
-rw-r--r--erts/emulator/Makefile.in13
-rw-r--r--erts/emulator/beam/bif.tab6
-rw-r--r--erts/emulator/beam/break.c1
-rw-r--r--erts/emulator/beam/dist.c16
-rw-r--r--erts/emulator/beam/erl_alloc.types1
-rwxr-xr-xerts/emulator/beam/erl_bif_info.c68
-rw-r--r--erts/emulator/beam/erl_bif_port.c173
-rw-r--r--erts/emulator/beam/erl_monitors.c20
-rw-r--r--erts/emulator/beam/erl_monitors.h2
-rw-r--r--erts/emulator/beam/erl_node_tables.c6
-rw-r--r--erts/emulator/beam/erl_port.h22
-rw-r--r--erts/emulator/beam/erl_port_task.c28
-rw-r--r--erts/emulator/beam/erl_port_task.h2
-rw-r--r--erts/emulator/beam/erl_process.c2
-rw-r--r--erts/emulator/beam/erl_process.h3
-rw-r--r--erts/emulator/beam/erl_process_dump.c37
-rw-r--r--erts/emulator/beam/io.c315
-rw-r--r--erts/emulator/drivers/common/inet_drv.c131
-rwxr-xr-xerts/emulator/sys/win32/sys.c26
-rw-r--r--erts/emulator/test/busy_port_SUITE.erl2
20 files changed, 440 insertions, 434 deletions
diff --git a/erts/emulator/Makefile.in b/erts/emulator/Makefile.in
index 4f2f647742..2031ec3a4c 100644
--- a/erts/emulator/Makefile.in
+++ b/erts/emulator/Makefile.in
@@ -48,6 +48,7 @@ CREATE_DIRS=
LDFLAGS=@LDFLAGS@
ARFLAGS=rc
+OMIT_OMIT_FP=no
ifeq ($(TYPE),debug)
PURIFY =
@@ -113,6 +114,13 @@ TYPEMARKER = .lcnt
TYPE_FLAGS = @CFLAGS@ -DERTS_ENABLE_LOCK_COUNT
else
+ifeq ($(TYPE),frmptr)
+PURIFY =
+OMIT_OMIT_FP=yes
+TYPEMARKER = .frmptr
+TYPE_FLAGS = @CFLAGS@ -DERTS_FRMPTR
+else
+
# If type isn't one of the above, it *is* opt type...
override TYPE=opt
PURIFY =
@@ -126,6 +134,7 @@ endif
endif
endif
endif
+endif
#
# NOTE: When adding a new type update ERL_BUILD_TYPE_MARKER in sys/unix/sys.c
@@ -219,8 +228,6 @@ LIB_PREFIX=lib
LIB_SUFFIX=.a
endif
-OMIT_OMIT_FP=no
-
ifeq (@EMU_LOCK_CHECKING@,yes)
NO_INLINE_FUNCTIONS=true
endif
@@ -238,7 +245,7 @@ ifeq ($(NO_INLINE_FUNCTIONS),true)
GEN_OPT_FLGS = $(OPT_LEVEL) -fno-inline-functions
else
ifeq ($(OMIT_OMIT_FP),yes)
-GEN_OPT_FLGS = $(OPT_LEVEL)
+GEN_OPT_FLGS = $(OPT_LEVEL) -fno-omit-frame-pointer
else
GEN_OPT_FLGS = $(OPT_LEVEL) -fomit-frame-pointer
endif
diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 8bc994c8c3..dc8e9101de 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -154,8 +154,10 @@ bif erts_internal:port_command/3
bif erts_internal:port_control/3
bif erts_internal:port_close/1
bif erts_internal:port_connect/2
-bif erts_internal:port_set_data/2
-bif erts_internal:port_get_data/1
+
+# inet_db support
+bif erlang:port_set_data/2
+bif erlang:port_get_data/1
# Tracing & debugging.
bif erlang:trace_pattern/2
diff --git a/erts/emulator/beam/break.c b/erts/emulator/beam/break.c
index 9aa1e5f30d..e34fc8388c 100644
--- a/erts/emulator/beam/break.c
+++ b/erts/emulator/beam/break.c
@@ -331,6 +331,7 @@ print_process_info(int to, void *to_arg, Process *p)
erts_print(to, to_arg, "Heap unused: %bpu\n", (p->hend - p->htop));
erts_print(to, to_arg, "OldHeap unused: %bpu\n",
(OLD_HEAP(p) == NULL) ? 0 : (OLD_HEND(p) - OLD_HTOP(p)) );
+ erts_print(to, to_arg, "Memory: %beu\n", erts_process_memory(p));
if (garbing) {
print_garb_info(to, to_arg, p);
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 0781665f05..44f4eb9d43 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -455,7 +455,7 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
Eterm nd_reason = (reason == am_no_network
? am_no_network
: am_net_kernel_terminated);
- erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
for (tdep = erts_hidden_dist_entries; tdep; tdep = tdep->next)
no_dist_port++;
@@ -464,7 +464,7 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
/* KILL all port controllers */
if (no_dist_port == 0)
- erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
else {
Eterm def_buf[128];
int i = 0;
@@ -483,7 +483,7 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
ASSERT(is_internal_port(tdep->cid));
dist_port[i++] = tdep->cid;
}
- erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
for (i = 0; i < no_dist_port; i++) {
Port *prt = erts_port_lookup(dist_port[i],
@@ -2560,9 +2560,9 @@ BIF_RETTYPE setnode_2(BIF_ALIST_2)
erts_smp_proc_unlock(net_kernel, ERTS_PROC_LOCK_MAIN);
#ifdef DEBUG
- erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
ASSERT(!erts_visible_dist_entries && !erts_hidden_dist_entries);
- erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
#endif
erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
@@ -2912,7 +2912,7 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
length = 0;
- erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
ASSERT(erts_no_of_not_connected_dist_entries >= 0);
ASSERT(erts_no_of_hidden_dist_entries >= 0);
@@ -2929,7 +2929,7 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
result = NIL;
if (length == 0) {
- erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
goto done;
}
@@ -2958,7 +2958,7 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
hp += 2;
}
ASSERT(endp == hp);
- erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+ erts_smp_rwmtx_runlock(&erts_dist_table_rwmtx);
done:
UnUseTmpHeap(2,BIF_P);
diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 5e3615ccc2..5a92ab7f24 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -272,6 +272,7 @@ type CODE_IX_LOCK_Q SHORT_LIVED SYSTEM code_ix_lock_q
type PROC_INTERVAL LONG_LIVED SYSTEM process_interval
type BUSY_CALLER_TAB SHORT_LIVED SYSTEM busy_caller_table
type BUSY_CALLER SHORT_LIVED SYSTEM busy_caller
+type PORT_DATA_HEAP STANDARD SYSTEM port_data_heap
+if threads_no_smp
# Need thread safe allocs, but std_alloc and fix_alloc are not;
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 379d3eecc6..54eefe8d12 100755
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -114,6 +114,9 @@ static char erts_system_version[] = ("Erlang " ERLANG_OTP_RELEASE
#ifdef VALGRIND
" [valgrind-compiled]"
#endif
+#ifdef ERTS_FRMPTR
+ " [frame-pointer]"
+#endif
#ifdef USE_DTRACE
" [dtrace]"
#endif
@@ -486,27 +489,6 @@ collect_one_suspend_monitor(ErtsSuspendMonitor *smon, void *vsmicp)
}
}
-
-static void one_link_size(ErtsLink *lnk, void *vpu)
-{
- Uint *pu = vpu;
- *pu += ERTS_LINK_SIZE*sizeof(Uint);
- if(!IS_CONST(lnk->pid))
- *pu += NC_HEAP_SIZE(lnk->pid)*sizeof(Uint);
- if (lnk->type != LINK_NODE && ERTS_LINK_ROOT(lnk) != NULL) {
- erts_doforall_links(ERTS_LINK_ROOT(lnk),&one_link_size,vpu);
- }
-}
-static void one_mon_size(ErtsMonitor *mon, void *vpu)
-{
- Uint *pu = vpu;
- *pu += ERTS_MONITOR_SIZE*sizeof(Uint);
- if(!IS_CONST(mon->pid))
- *pu += NC_HEAP_SIZE(mon->pid)*sizeof(Uint);
- if(!IS_CONST(mon->ref))
- *pu += NC_HEAP_SIZE(mon->ref)*sizeof(Uint);
-}
-
/*
* process_info/[1,2]
*/
@@ -1420,41 +1402,8 @@ process_info_aux(Process *BIF_P,
}
case am_memory: { /* Memory consumed in bytes */
- ErlMessage *mp;
- Uint size = 0;
Uint hsz = 3;
- struct saved_calls *scb;
- size += sizeof(Process);
-
- ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
-
- erts_doforall_links(ERTS_P_LINKS(rp), &one_link_size, &size);
- erts_doforall_monitors(ERTS_P_MONITORS(rp), &one_mon_size, &size);
- size += (rp->heap_sz + rp->mbuf_sz) * sizeof(Eterm);
- if (rp->old_hend && rp->old_heap)
- size += (rp->old_hend - rp->old_heap) * sizeof(Eterm);
-
- size += rp->msg.len * sizeof(ErlMessage);
-
- for (mp = rp->msg.first; mp; mp = mp->next)
- if (mp->data.attached)
- size += erts_msg_attached_data_size(mp)*sizeof(Eterm);
-
- if (rp->arg_reg != rp->def_arg_reg) {
- size += rp->arity * sizeof(rp->arg_reg[0]);
- }
-
- if (rp->psd)
- size += sizeof(ErtsPSD);
-
- scb = ERTS_PROC_GET_SAVED_CALLS_BUF(rp);
- if (scb) {
- size += (sizeof(struct saved_calls)
- + (scb->len-1) * sizeof(scb->ct[0]));
- }
-
- size += erts_dicts_mem_size(rp);
-
+ Uint size = erts_process_memory(rp);
(void) erts_bld_uint(NULL, &hsz, size);
hp = HAlloc(BIF_P, hsz);
res = erts_bld_uint(&hp, NULL, size);
@@ -2096,6 +2045,9 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
#elif defined(ERTS_ENABLE_LOCK_COUNT)
ERTS_DECL_AM(lcnt);
BIF_RET(AM_lcnt);
+#elif defined(ERTS_FRMPTR)
+ ERTS_DECL_AM(frmptr);
+ BIF_RET(AM_frmptr);
#else
BIF_RET(am_opt);
#endif
@@ -2861,12 +2813,10 @@ erts_bld_port_info(Eterm **hpp, ErlOffHeap *ohp, Uint *szp, Port *prt, Eterm ite
included though).
*/
Uint size = 0;
- ErlHeapFragment* bp;
- erts_doforall_links(ERTS_P_LINKS(prt), &one_link_size, &size);
+ erts_doforall_links(ERTS_P_LINKS(prt), &erts_one_link_size, &size);
- for (bp = prt->bp; bp; bp = bp->next)
- size += sizeof(ErlHeapFragment) + (bp->alloc_size - 1)*sizeof(Eterm);
+ size += erts_port_data_size(prt);
if (prt->linebuf)
size += sizeof(LineBuf) + prt->linebuf->ovsiz;
diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c
index 44fa41c7b6..109c54fd7f 100644
--- a/erts/emulator/beam/erl_bif_port.c
+++ b/erts/emulator/beam/erl_bif_port.c
@@ -423,57 +423,164 @@ BIF_RETTYPE erts_internal_port_info_2(BIF_ALIST_2)
}
}
+/*
+ * The erlang:port_set_data()/erlang:port_get_data() operations should
+ * be viewed as operations on a table (inet_db) with data values
+ * associated with port identifier keys. That is, these operations are
+ * *not* signals to/from ports.
+ */
+
+#if (TAG_PRIMARY_IMMED1 & 0x3) == 0
+# error "erlang:port_set_data()/erlang:port_get_data() needs to be rewritten!"
+#endif
+
+typedef struct {
+ ErtsThrPrgrLaterOp later_op;
+ Uint hsize;
+ Eterm data;
+ ErlOffHeap off_heap;
+ Eterm heap[1];
+} ErtsPortDataHeap;
-BIF_RETTYPE erts_internal_port_set_data_2(BIF_ALIST_2)
+static void
+free_port_data_heap(void *vpdhp)
{
- Eterm ref;
+ erts_cleanup_offheap(&((ErtsPortDataHeap *) vpdhp)->off_heap);
+ erts_free(ERTS_ALC_T_PORT_DATA_HEAP, vpdhp);
+}
+
+static ERTS_INLINE void
+cleanup_old_port_data(erts_aint_t data)
+{
+ if ((data & 0x3) != 0) {
+ ASSERT(is_immed((Eterm) data));
+ }
+ else {
+#ifdef ERTS_SMP
+ ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
+ size_t size;
+ ERTS_SMP_DATA_DEPENDENCY_READ_MEMORY_BARRIER;
+ size = sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm) - 1);
+ erts_schedule_thr_prgr_later_cleanup_op(free_port_data_heap,
+ (void *) pdhp,
+ &pdhp->later_op,
+ size);
+#else
+ free_port_data_heap((void *) data);
+#endif
+ }
+}
+
+void
+erts_init_port_data(Port *prt)
+{
+ erts_smp_atomic_init_nob(&prt->data, (erts_aint_t) am_undefined);
+}
+
+void
+erts_cleanup_port_data(Port *prt)
+{
+ ASSERT(erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_INVALID_LOOKUP);
+ cleanup_old_port_data(erts_smp_atomic_read_nob(&prt->data));
+ erts_smp_atomic_set_nob(&prt->data, (erts_aint_t) THE_NON_VALUE);
+}
+
+Uint
+erts_port_data_size(Port *prt)
+{
+ erts_aint_t data = erts_smp_atomic_read_ddrb(&prt->data);
+
+ if ((data & 0x3) != 0) {
+ ASSERT(is_immed((Eterm) (UWord) data));
+ return (Uint) 0;
+ }
+ else {
+ ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
+ return (Uint) sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm)-1);
+ }
+}
+
+ErlOffHeap *
+erts_port_data_offheap(Port *prt)
+{
+ erts_aint_t data = erts_smp_atomic_read_ddrb(&prt->data);
+
+ if ((data & 0x3) != 0) {
+ ASSERT(is_immed((Eterm) (UWord) data));
+ return NULL;
+ }
+ else {
+ ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
+ return &pdhp->off_heap;
+ }
+}
+
+BIF_RETTYPE port_set_data_2(BIF_ALIST_2)
+{
+ /*
+ * This is not a signal. See comment above.
+ */
+ erts_aint_t data;
Port* prt;
prt = lookup_port(BIF_P, BIF_ARG_1);
if (!prt)
- BIF_RET(am_badarg);
+ BIF_ERROR(BIF_P, BADARG);
- switch (erts_port_set_data(BIF_P, prt, BIF_ARG_2, &ref)) {
- case ERTS_PORT_OP_CALLER_EXIT:
- case ERTS_PORT_OP_BADARG:
- case ERTS_PORT_OP_DROPPED:
- BIF_RET(am_badarg);
- case ERTS_PORT_OP_SCHEDULED:
- ASSERT(is_internal_ref(ref));
- BIF_RET(ref);
- case ERTS_PORT_OP_DONE:
- BIF_RET(am_true);
- default:
- ERTS_INTERNAL_ERROR("Unexpected erts_port_set_data() result");
- BIF_RET(am_internal_error);
+ if (is_immed(BIF_ARG_2)) {
+ data = (erts_aint_t) BIF_ARG_2;
+ ASSERT((data & 0x3) != 0);
}
+ else {
+ ErtsPortDataHeap *pdhp;
+ Uint hsize;
+ Eterm *hp;
+
+ hsize = size_object(BIF_ARG_2);
+ pdhp = erts_alloc(ERTS_ALC_T_PORT_DATA_HEAP,
+ sizeof(ErtsPortDataHeap) + hsize*(sizeof(Eterm)-1));
+ hp = &pdhp->heap[0];
+ pdhp->off_heap.first = NULL;
+ pdhp->off_heap.overhead = 0;
+ pdhp->data = copy_struct(BIF_ARG_2, hsize, &hp, &pdhp->off_heap);
+ data = (erts_aint_t) pdhp;
+ ASSERT((data & 0x3) == 0);
+ }
+
+ data = erts_smp_atomic_xchg_wb(&prt->data, data);
+
+ cleanup_old_port_data(data);
+
+ BIF_RET(am_true);
}
-BIF_RETTYPE erts_internal_port_get_data_1(BIF_ALIST_1)
+BIF_RETTYPE port_get_data_1(BIF_ALIST_1)
{
- Eterm retval;
+ /*
+ * This is not a signal. See comment above.
+ */
+ Eterm res;
+ erts_aint_t data;
Port* prt;
prt = lookup_port(BIF_P, BIF_ARG_1);
if (!prt)
- BIF_RET(am_badarg);
+ BIF_ERROR(BIF_P, BADARG);
- switch (erts_port_get_data(BIF_P, prt, &retval)) {
- case ERTS_PORT_OP_CALLER_EXIT:
- case ERTS_PORT_OP_BADARG:
- case ERTS_PORT_OP_DROPPED:
- BIF_RET(am_badarg);
- case ERTS_PORT_OP_SCHEDULED:
- ASSERT(is_internal_ref(retval));
- BIF_RET(retval);
- case ERTS_PORT_OP_DONE:
- ASSERT(is_not_internal_ref(retval));
- BIF_RET(retval);
- default:
- ERTS_INTERNAL_ERROR("Unexpected erts_port_get_data() result");
- BIF_RET(am_internal_error);
+ data = erts_smp_atomic_read_ddrb(&prt->data);
+
+ if ((data & 0x3) != 0) {
+ res = (Eterm) (UWord) data;
+ ASSERT(is_immed(res));
+ }
+ else {
+ ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
+ Eterm *hp = HAlloc(BIF_P, pdhp->hsize);
+ res = copy_struct(pdhp->data, pdhp->hsize, &hp, &MSO(BIF_P));
}
+
+ BIF_RET(res);
}
/*
diff --git a/erts/emulator/beam/erl_monitors.c b/erts/emulator/beam/erl_monitors.c
index 70e592cc5f..244a2b26db 100644
--- a/erts/emulator/beam/erl_monitors.c
+++ b/erts/emulator/beam/erl_monitors.c
@@ -1024,3 +1024,23 @@ Eterm erts_debug_dump_links_1(BIF_ALIST_1)
}
}
}
+
+void erts_one_link_size(ErtsLink *lnk, void *vpu)
+{
+ Uint *pu = vpu;
+ *pu += ERTS_LINK_SIZE*sizeof(Uint);
+ if(!IS_CONST(lnk->pid))
+ *pu += NC_HEAP_SIZE(lnk->pid)*sizeof(Uint);
+ if (lnk->type != LINK_NODE && ERTS_LINK_ROOT(lnk) != NULL) {
+ erts_doforall_links(ERTS_LINK_ROOT(lnk),&erts_one_link_size,vpu);
+ }
+}
+void erts_one_mon_size(ErtsMonitor *mon, void *vpu)
+{
+ Uint *pu = vpu;
+ *pu += ERTS_MONITOR_SIZE*sizeof(Uint);
+ if(!IS_CONST(mon->pid))
+ *pu += NC_HEAP_SIZE(mon->pid)*sizeof(Uint);
+ if(!IS_CONST(mon->ref))
+ *pu += NC_HEAP_SIZE(mon->ref)*sizeof(Uint);
+}
diff --git a/erts/emulator/beam/erl_monitors.h b/erts/emulator/beam/erl_monitors.h
index 6a360a2336..fb11dbbd22 100644
--- a/erts/emulator/beam/erl_monitors.h
+++ b/erts/emulator/beam/erl_monitors.h
@@ -170,6 +170,8 @@ ErtsSuspendMonitor *erts_lookup_suspend_monitor(ErtsSuspendMonitor *root,
Eterm pid);
void erts_delete_suspend_monitor(ErtsSuspendMonitor **root, Eterm pid);
void erts_init_monitors(void);
+void erts_one_link_size(ErtsLink *lnk, void *vpu);
+void erts_one_mon_size(ErtsMonitor *mon, void *vpu);
#define erts_doforall_monitors erts_sweep_monitors
#define erts_doforall_links erts_sweep_links
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index ebfba065d1..e688e55c88 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -1371,6 +1371,7 @@ setup_reference_table(void)
/* Insert all ports */
max = erts_ptab_max(&erts_port);
for (i = 0; i < max; i++) {
+ ErlOffHeap *ohp;
erts_aint32_t state;
Port *prt;
@@ -1389,8 +1390,9 @@ setup_reference_table(void)
if (ERTS_P_MONITORS(prt))
insert_monitors(ERTS_P_MONITORS(prt), prt->common.id);
/* Insert port data */
- for(hfp = prt->bp; hfp; hfp = hfp->next)
- insert_offheap(&(hfp->off_heap), HEAP_REF, prt->common.id);
+ ohp = erts_port_data_offheap(prt);
+ if (ohp)
+ insert_offheap(ohp, HEAP_REF, prt->common.id);
/* Insert controller */
if (prt->dist_entry)
insert_dist_entry(prt->dist_entry,
diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index 377aa72ed5..ad3f104a68 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -167,8 +167,7 @@ struct _erl_drv_port {
#endif
erts_atomic_t connected; /* A connected process */
Eterm caller; /* Current caller. */
- Eterm data; /* Data associated with port. */
- ErlHeapFragment* bp; /* Heap fragment holding data (NULL if imm data). */
+ erts_smp_atomic_t data; /* Data associated with port. */
Uint bytes_in; /* Number of bytes read */
Uint bytes_out; /* Number of bytes written */
@@ -189,6 +188,12 @@ struct _erl_drv_port {
int reds; /* Only used while executing driver callbacks */
};
+
+void erts_init_port_data(Port *);
+void erts_cleanup_port_data(Port *);
+Uint erts_port_data_size(Port *);
+ErlOffHeap *erts_port_data_offheap(Port *);
+
#define ERTS_PORT_GET_CONNECTED(PRT) \
((Eterm) erts_atomic_read_nob(&(PRT)->connected))
#define ERTS_PORT_SET_CONNECTED(PRT, PID) \
@@ -326,8 +331,6 @@ extern erts_smp_atomic_t erts_bytes_in; /* no bytes sent into the system */
#define ERTS_PORT_REDS_CONTROL (CONTEXT_REDS/100)
#define ERTS_PORT_REDS_CALL (CONTEXT_REDS/50)
#define ERTS_PORT_REDS_INFO (CONTEXT_REDS/100)
-#define ERTS_PORT_REDS_SET_DATA (CONTEXT_REDS/100)
-#define ERTS_PORT_REDS_GET_DATA (CONTEXT_REDS/100)
#define ERTS_PORT_REDS_TERMINATE (CONTEXT_REDS/50)
void print_port_info(Port *, int, void *);
@@ -794,8 +797,6 @@ struct binary;
#define ERTS_P2P_SIG_TYPE_INFO 7
#define ERTS_P2P_SIG_TYPE_LINK 8
#define ERTS_P2P_SIG_TYPE_UNLINK 9
-#define ERTS_P2P_SIG_TYPE_SET_DATA 10
-#define ERTS_P2P_SIG_TYPE_GET_DATA 11
#define ERTS_P2P_SIG_TYPE_BITS 4
#define ERTS_P2P_SIG_TYPE_MASK \
@@ -810,6 +811,7 @@ struct binary;
#define ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT ERTS_P2P_SIG_DATA_FLG(4)
#define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK ERTS_P2P_SIG_DATA_FLG(5)
#define ERTS_P2P_SIG_DATA_FLG_SCHED ERTS_P2P_SIG_DATA_FLG(6)
+#define ERTS_P2P_SIG_DATA_FLG_ASYNC ERTS_P2P_SIG_DATA_FLG(7)
struct ErtsProc2PortSigData_ {
int flags;
@@ -856,10 +858,6 @@ struct ErtsProc2PortSigData_ {
struct {
Eterm from;
} unlink;
- struct {
- ErlHeapFragment *bp;
- Eterm data;
- } set_data;
} u;
} ;
@@ -919,6 +917,7 @@ erts_schedule_proc2port_signal(Process *,
Eterm *,
ErtsProc2PortSigData *,
int,
+ ErtsPortTaskHandle *,
ErtsProc2PortSigCallback);
int erts_deliver_port_exit(Port *, Eterm, Eterm, int);
@@ -932,6 +931,7 @@ int erts_deliver_port_exit(Port *, Eterm, Eterm, int);
#define ERTS_PORT_SIG_FLG_BROKEN_LINK ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK
#define ERTS_PORT_SIG_FLG_BAD_OUTPUT ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT
#define ERTS_PORT_SIG_FLG_FORCE_SCHED ERTS_P2P_SIG_DATA_FLG_SCHED
+#define ERTS_PORT_SIG_FLG_ASYNC ERTS_P2P_SIG_DATA_FLG_ASYNC
/* ERTS_PORT_SIG_FLG_FORCE_IMM_CALL only when crash dumping... */
#define ERTS_PORT_SIG_FLG_FORCE_IMM_CALL ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT
@@ -953,7 +953,5 @@ ErtsPortOpResult erts_port_unlink(Process *, Port *, Eterm, Eterm *);
ErtsPortOpResult erts_port_control(Process *, Port *, unsigned int, Eterm, Eterm *);
ErtsPortOpResult erts_port_call(Process *, Port *, unsigned int, Eterm, Eterm *);
ErtsPortOpResult erts_port_info(Process *, Port *, Eterm, Eterm *);
-ErtsPortOpResult erts_port_set_data(Process *, Port *, Eterm, Eterm *);
-ErtsPortOpResult erts_port_get_data(Process *, Port *, Eterm *);
#endif
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 0ed08bee01..53cb01a8c6 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -552,6 +552,19 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
}
}
+static ERTS_INLINE void
+set_tmp_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
+{
+ ptp->u.alive.handle = NULL;
+ if (pthp) {
+ /*
+ * IMPORTANT! Task either need to be aborted, or task handle
+ * need to be detached before thread progress has been made.
+ */
+ erts_smp_atomic_set_relb(pthp, (erts_aint_t) ptp);
+ }
+}
+
/*
* Busy port queue management
@@ -1182,6 +1195,13 @@ erl_drv_consume_timeslice(ErlDrvPort dprt, int percent)
return 1;
}
+void
+erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *pthp)
+{
+ ERTS_SMP_LC_ASSERT(erts_thr_progress_lc_is_delaying());
+ reset_port_task_handle(pthp);
+}
+
/*
* Abort a scheduled task.
*/
@@ -1206,7 +1226,7 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
ERTS_SMP_READ_MEMORY_BARRIER;
old_state = erts_smp_atomic32_read_nob(&ptp->state);
if (old_state == ERTS_PT_STATE_SCHEDULED) {
- ASSERT(saved_pthp == pthp);
+ ASSERT(!saved_pthp || saved_pthp == pthp);
}
#endif
@@ -1227,9 +1247,6 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
&erts_port_task_outstanding_io_tasks) > 0);
erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks);
break;
- case ERTS_PORT_TASK_PROC_SIG:
- ERTS_INTERNAL_ERROR("Aborted process to port signal");
- break;
default:
break;
}
@@ -1404,7 +1421,6 @@ erts_port_task_schedule(Eterm id,
}
case ERTS_PORT_TASK_PROC_SIG: {
va_list argp;
- ASSERT(!pthp);
va_start(argp, type);
sigdp = va_arg(argp, ErtsProc2PortSigData *);
ptp = p2p_sig_data_to_task(sigdp);
@@ -1412,7 +1428,7 @@ erts_port_task_schedule(Eterm id,
ptp->u.alive.flags |= va_arg(argp, int);
va_end(argp);
if (!(ptp->u.alive.flags & ERTS_PT_FLG_NOSUSPEND))
- set_handle(ptp, pthp);
+ set_tmp_handle(ptp, pthp);
else {
ns_pthlp = erts_alloc(ERTS_ALC_T_PT_HNDL_LIST,
sizeof(ErtsPortTaskHandleList));
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index ae6cd69ae2..d35a15c27b 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -243,7 +243,9 @@ int erts_port_task_execute(ErtsRunQueue *, Port **);
void erts_port_task_init(void);
#endif
+void erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *);
int erts_port_task_abort(ErtsPortTaskHandle *);
+
void erts_port_task_abort_nosuspend_tasks(Port *);
int erts_port_task_schedule(Eterm,
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 5d9c5af55f..7415a5721f 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -4670,6 +4670,8 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online)
esdp->reductions = 0;
init_sched_wall_time(&esdp->sched_wall_time);
+
+ erts_port_task_handle_init(&esdp->nosuspend_port_task_handle);
}
init_misc_aux_work();
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 3d3579fa7e..5a1f6bbe8d 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -507,6 +507,7 @@ struct ErtsSchedulerData_ {
Uint64 reductions;
ErtsSchedWallTime sched_wall_time;
+ ErtsPortTaskHandle nosuspend_port_task_handle;
#ifdef ERTS_DO_VERIFY_UNUSED_TEMP_ALLOC
erts_alloc_verify_func_t verify_unused_temp_alloc;
@@ -1421,6 +1422,8 @@ Eterm erts_debug_reader_groups_map(Process *c_p, int groups);
Uint erts_debug_nbalance(void);
int erts_debug_wait_deallocations(Process *c_p);
+Uint erts_process_memory(Process *c_p);
+
#ifdef ERTS_SMP
# define ERTS_GET_SCHEDULER_DATA_FROM_PROC(PROC) ((PROC)->scheduler_data)
# define ERTS_PROC_GET_SCHDATA(PROC) ((PROC)->scheduler_data)
diff --git a/erts/emulator/beam/erl_process_dump.c b/erts/emulator/beam/erl_process_dump.c
index a93229c473..6cd0d23b97 100644
--- a/erts/emulator/beam/erl_process_dump.c
+++ b/erts/emulator/beam/erl_process_dump.c
@@ -76,6 +76,43 @@ erts_deep_process_dump(int to, void *to_arg)
dump_binaries(to, to_arg, all_binaries);
}
+Uint erts_process_memory(Process *p) {
+ ErlMessage *mp;
+ Uint size = 0;
+ struct saved_calls *scb;
+ size += sizeof(Process);
+
+ ERTS_SMP_MSGQ_MV_INQ2PRIVQ(p);
+
+ erts_doforall_links(ERTS_P_LINKS(p), &erts_one_link_size, &size);
+ erts_doforall_monitors(ERTS_P_MONITORS(p), &erts_one_mon_size, &size);
+ size += (p->heap_sz + p->mbuf_sz) * sizeof(Eterm);
+ if (p->old_hend && p->old_heap)
+ size += (p->old_hend - p->old_heap) * sizeof(Eterm);
+
+ size += p->msg.len * sizeof(ErlMessage);
+
+ for (mp = p->msg.first; mp; mp = mp->next)
+ if (mp->data.attached)
+ size += erts_msg_attached_data_size(mp)*sizeof(Eterm);
+
+ if (p->arg_reg != p->def_arg_reg) {
+ size += p->arity * sizeof(p->arg_reg[0]);
+ }
+
+ if (p->psd)
+ size += sizeof(ErtsPSD);
+
+ scb = ERTS_PROC_GET_SAVED_CALLS_BUF(p);
+ if (scb) {
+ size += (sizeof(struct saved_calls)
+ + (scb->len-1) * sizeof(scb->ct[0]));
+ }
+
+ size += erts_dicts_mem_size(p);
+ return size;
+}
+
static void
dump_process_info(int to, void *to_arg, Process *p)
{
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index f93b599632..b6b7b47bd6 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -364,9 +364,8 @@ static Port *create_port(char *name,
ERTS_P_LINKS(prt) = NULL;
ERTS_P_MONITORS(prt) = NULL;
prt->linebuf = NULL;
- prt->bp = NULL;
prt->suspended = NULL;
- prt->data = am_undefined;
+ erts_init_port_data(prt);
prt->port_data_lock = NULL;
prt->control_flags = 0;
prt->bytes_in = 0;
@@ -1442,6 +1441,7 @@ erts_schedule_proc2port_signal(Process *c_p,
Eterm *refp,
ErtsProc2PortSigData *sigdp,
int task_flags,
+ ErtsPortTaskHandle *pthp,
ErtsProc2PortSigCallback callback)
{
int sched_res;
@@ -1491,7 +1491,7 @@ erts_schedule_proc2port_signal(Process *c_p,
/* Schedule port close call for later execution... */
sched_res = erts_port_task_schedule(prt->common.id,
- NULL,
+ pthp,
ERTS_PORT_TASK_PROC_SIG,
sigdp,
callback,
@@ -1629,6 +1629,7 @@ bad_port_signal(Process *c_p,
refp,
sigdp,
0,
+ NULL,
port_badsig);
}
@@ -1838,8 +1839,11 @@ erts_port_output(Process *c_p,
ErlIOVec *evp = NULL;
char *buf = NULL;
int force_immediate_call = (flags & ERTS_PORT_SIG_FLG_FORCE_IMM_CALL);
+ int async_nosuspend;
+ ErtsPortTaskHandle *ns_pthp;
ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+ | ERTS_PORT_SIG_FLG_ASYNC
| ERTS_PORT_SIG_FLG_NOSUSPEND
| ERTS_PORT_SIG_FLG_FORCE
| ERTS_PORT_SIG_FLG_FORCE_IMM_CALL)) == 0);
@@ -1861,6 +1865,12 @@ erts_port_output(Process *c_p,
? ERTS_PORT_OP_DROPPED
: ERTS_PORT_OP_BUSY);
+ async_nosuspend = ((flags & (ERTS_PORT_SIG_FLG_ASYNC
+ | ERTS_PORT_SIG_FLG_NOSUSPEND
+ | ERTS_PORT_SIG_FLG_FORCE))
+ == (ERTS_PORT_SIG_FLG_ASYNC
+ | ERTS_PORT_SIG_FLG_NOSUSPEND));
+
try_call = (force_immediate_call /* crash dumping */
|| !(sched_flags & (invalid_flags
| ERTS_PTS_FLGS_FORCE_SCHEDULE_OP)));
@@ -1995,6 +2005,15 @@ erts_port_output(Process *c_p,
return ERTS_PORT_OP_DONE;
case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
sched_flags = try_call_state.sched_flags;
+ if (async_nosuspend
+ && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) {
+ driver_free_binary(cbin);
+ if (evp != &ev)
+ erts_free(ERTS_ALC_T_TMP, evp);
+ return ((sched_flags & ERTS_PTS_FLG_EXIT)
+ ? ERTS_PORT_OP_DROPPED
+ : ERTS_PORT_OP_BUSY);
+ }
case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
/* Schedule outputv() call instead... */
break;
@@ -2142,6 +2161,13 @@ erts_port_output(Process *c_p,
return ERTS_PORT_OP_DONE;
case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
sched_flags = try_call_state.sched_flags;
+ if (async_nosuspend
+ && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) {
+ erts_free(ERTS_ALC_T_TMP, buf);
+ return ((sched_flags & ERTS_PTS_FLG_EXIT)
+ ? ERTS_PORT_OP_DROPPED
+ : ERTS_PORT_OP_BUSY);
+ }
case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
/* Schedule outputv() call instead... */
break;
@@ -2163,20 +2189,33 @@ erts_port_output(Process *c_p,
task_flags = ERTS_PT_FLG_WAIT_BUSY;
sigdp->flags |= flags;
+ ns_pthp = NULL;
if (flags & (ERTS_P2P_SIG_DATA_FLG_FORCE|ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)) {
task_flags = 0;
if (flags & ERTS_P2P_SIG_DATA_FLG_FORCE)
sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND;
+ else if (async_nosuspend) {
+ ErtsSchedulerData *esdp = (c_p
+ ? ERTS_PROC_GET_SCHDATA(c_p)
+ : erts_get_scheduler_data());
+ ASSERT(esdp);
+ ns_pthp = &esdp->nosuspend_port_task_handle;
+ sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND;
+ }
else if (flags & ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)
task_flags = ERTS_PT_FLG_NOSUSPEND;
}
+ ASSERT(ns_pthp || !async_nosuspend);
+ ASSERT(async_nosuspend || !ns_pthp);
+
res = erts_schedule_proc2port_signal(c_p,
prt,
c_p ? c_p->common.id : ERTS_INVALID_PID,
refp,
sigdp,
task_flags,
+ ns_pthp,
port_sig_callback);
if (res != ERTS_PORT_OP_SCHEDULED) {
@@ -2187,9 +2226,23 @@ erts_port_output(Process *c_p,
return res;
}
- if (!(sched_flags & ERTS_PTS_FLG_EXIT) && (sched_flags & busy_flgs))
- return ERTS_PORT_OP_BUSY_SCHEDULED;
-
+ if (!(flags & ERTS_PORT_SIG_FLG_FORCE)) {
+ sched_flags = erts_smp_atomic32_read_acqb(&prt->sched.flags);
+ if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT)) {
+ if (async_nosuspend)
+ erts_port_task_tmp_handle_detach(ns_pthp);
+ }
+ else {
+ if (!async_nosuspend)
+ return ERTS_PORT_OP_BUSY_SCHEDULED;
+ else {
+ if (erts_port_task_abort(ns_pthp) == 0)
+ return ERTS_PORT_OP_BUSY;
+ else
+ erts_port_task_tmp_handle_detach(ns_pthp);
+ }
+ }
+ }
return res;
bad_value:
@@ -2285,6 +2338,7 @@ erts_port_exit(Process *c_p,
ErlHeapFragment *bp = NULL;
ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+ | ERTS_PORT_SIG_FLG_ASYNC
| ERTS_PORT_SIG_FLG_BROKEN_LINK
| ERTS_PORT_SIG_FLG_FORCE_SCHED)) == 0);
@@ -2345,6 +2399,7 @@ erts_port_exit(Process *c_p,
refp,
sigdp,
0,
+ NULL,
port_sig_exit);
if (res == ERTS_PORT_OP_DROPPED) {
@@ -2460,7 +2515,8 @@ erts_port_connect(Process *c_p,
!refp,
am_connect);
- ASSERT((flags & ~ERTS_PORT_SIG_FLG_BANG_OP) == 0);
+ ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP
+ | ERTS_PORT_SIG_FLG_ASYNC)) == 0);
if (is_not_internal_pid(connect))
connect_id = NIL; /* Fail in op (for signal order) */
@@ -2499,6 +2555,7 @@ erts_port_connect(Process *c_p,
refp,
sigdp,
0,
+ NULL,
port_sig_connect);
}
@@ -2555,6 +2612,7 @@ erts_port_unlink(Process *c_p, Port *prt, Eterm from, Eterm *refp)
refp,
sigdp,
0,
+ NULL,
port_sig_unlink);
}
@@ -2644,6 +2702,7 @@ erts_port_link(Process *c_p, Port *prt, Eterm to, Eterm *refp)
refp,
sigdp,
0,
+ NULL,
port_sig_link);
}
@@ -3371,11 +3430,8 @@ terminate_port(Port *prt)
erts_free(ERTS_ALC_T_LINEBUF, (void *) prt->linebuf);
prt->linebuf = NULL;
}
- if (prt->bp != NULL) {
- free_message_buffer(prt->bp);
- prt->bp = NULL;
- prt->data = am_undefined;
- }
+
+ erts_cleanup_port_data(prt);
if (prt->psd)
erts_free(ERTS_ALC_T_PRTSD, prt->psd);
@@ -3625,6 +3681,10 @@ erts_port_command(Process *c_p,
ASSERT(port);
flags |= ERTS_PORT_SIG_FLG_BANG_OP;
+ if (!erts_port_synchronous_ops) {
+ flags |= ERTS_PORT_SIG_FLG_ASYNC;
+ refp = NULL;
+ }
if (is_tuple_arity(command, 2)) {
Eterm cntd;
@@ -3632,21 +3692,14 @@ erts_port_command(Process *c_p,
cntd = tp[1];
if (is_internal_pid(cntd)) {
if (tp[2] == am_close) {
- if (!erts_port_synchronous_ops)
- refp = NULL;
flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
return erts_port_exit(c_p, flags, port, cntd, am_normal, refp);
} else if (is_tuple_arity(tp[2], 2)) {
tp = tuple_val(tp[2]);
if (tp[1] == am_command) {
- if (!(flags & ERTS_PORT_SIG_FLG_NOSUSPEND)
- && !erts_port_synchronous_ops)
- refp = NULL;
return erts_port_output(c_p, flags, port, cntd, tp[2], refp);
}
else if (tp[1] == am_connect) {
- if (!erts_port_synchronous_ops)
- refp = NULL;
flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
return erts_port_connect(c_p, flags, port, cntd, tp[2], refp);
}
@@ -3655,8 +3708,6 @@ erts_port_command(Process *c_p,
}
/* badsig */
- if (!erts_port_synchronous_ops)
- refp = NULL;
flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
return bad_port_signal(c_p, flags, port, c_p->common.id, refp, am_command);
}
@@ -4053,6 +4104,7 @@ erts_port_control(Process* c_p,
retvalp,
sigdp,
0,
+ NULL,
port_sig_control);
if (res != ERTS_PORT_OP_SCHEDULED) {
cleanup_scheduled_control(binp, bufp);
@@ -4333,6 +4385,7 @@ erts_port_call(Process* c_p,
retvalp,
sigdp,
0,
+ NULL,
port_sig_call);
if (res != ERTS_PORT_OP_SCHEDULED) {
cleanup_scheduled_call(bufp);
@@ -4499,228 +4552,10 @@ erts_port_info(Process* c_p,
retvalp,
sigdp,
0,
+ NULL,
port_sig_info);
}
-static int
-port_sig_set_data(Port *prt,
- erts_aint32_t state,
- int op,
- ErtsProc2PortSigData *sigdp)
-{
- ASSERT(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY);
-
- if (op == ERTS_PROC2PORT_SIG_EXEC) {
- if (prt->bp)
- free_message_buffer(prt->bp);
- prt->bp = sigdp->u.set_data.bp;
- prt->data = sigdp->u.set_data.data;
- port_sched_op_reply(sigdp->caller, sigdp->ref, am_true);
- }
- else {
- if (sigdp->u.set_data.bp)
- free_message_buffer(sigdp->u.set_data.bp);
- port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg);
- }
- return ERTS_PORT_REDS_SET_DATA;
-}
-
-ErtsPortOpResult
-erts_port_set_data(Process* c_p,
- Port *prt,
- Eterm data,
- Eterm *refp)
-{
- ErtsPortOpResult res;
- Eterm set_data;
- ErlHeapFragment *bp;
- ErtsProc2PortSigData *sigdp;
- ErtsTryImmDrvCallResult try_call_res;
- ErtsTryImmDrvCallState try_call_state
- = ERTS_INIT_TRY_IMM_DRV_CALL_STATE(
- c_p,
- prt,
- ERTS_PORT_SFLGS_INVALID_LOOKUP,
- 0,
- !refp,
- am_set_data);
-
- if (is_immed(data)) {
- set_data = data;
- bp = NULL;
- }
- else {
- Eterm *hp;
- Uint sz = size_object(data);
- bp = new_message_buffer(sz);
- hp = bp->mem;
- set_data = copy_struct(data, sz, &hp, &bp->off_heap);
- }
-
- try_call_res = try_imm_drv_call(&try_call_state);
- switch (try_call_res) {
- case ERTS_TRY_IMM_DRV_CALL_OK:
- if (prt->bp)
- free_message_buffer(prt->bp);
- prt->bp = bp;
- prt->data = set_data;
- finalize_imm_drv_call(&try_call_state);
- BUMP_REDS(c_p, ERTS_PORT_REDS_SET_DATA);
- return ERTS_PORT_OP_DONE;
- case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT:
- return ERTS_PORT_OP_DROPPED;
- case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
- case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
- /* Schedule call instead... */
- break;
- }
-
- sigdp = erts_port_task_alloc_p2p_sig_data();
- sigdp->flags = ERTS_P2P_SIG_TYPE_SET_DATA;
- sigdp->u.set_data.data = set_data;
- sigdp->u.set_data.bp = bp;
-
- res = erts_schedule_proc2port_signal(c_p,
- prt,
- c_p->common.id,
- refp,
- sigdp,
- 0,
- port_sig_set_data);
- if (res != ERTS_PORT_OP_SCHEDULED && bp)
- free_message_buffer(bp);
- return res;
-}
-
-static int
-port_sig_get_data(Port *prt,
- erts_aint32_t state,
- int op,
- ErtsProc2PortSigData *sigdp)
-{
- ASSERT(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY);
- if (op != ERTS_PROC2PORT_SIG_EXEC)
- port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg);
- else {
- Process *rp;
- ErtsProcLocks rp_locks = 0;
-
- rp = erts_proc_lookup_raw(sigdp->caller);
- if (rp) {
- Uint hsz;
- Eterm *hp, *hp_start;
- Eterm data, msg;
- ErlHeapFragment *bp;
- ErlOffHeap *ohp;
-
- hsz = ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE;
- hsz += 3;
- if (prt->bp)
- hsz += prt->bp->used_size;
-
- hp_start = hp = erts_alloc_message_heap(hsz,
- &bp,
- &ohp,
- rp,
- &rp_locks);
-
- if (is_immed(prt->data))
- data = prt->data;
- else
- data = copy_struct(prt->data,
- prt->bp->used_size,
- &hp,
- &bp->off_heap);
-
-
-
- msg = TUPLE2(hp, am_ok, data);
- hp += 3;
-
- queue_port_sched_op_reply(rp,
- &rp_locks,
- hp_start,
- hp,
- hsz,
- bp,
- sigdp->ref,
- msg);
- if (rp_locks)
- erts_smp_proc_unlock(rp, rp_locks);
- }
- }
- return ERTS_PORT_REDS_GET_DATA;
-}
-
-ErtsPortOpResult
-erts_port_get_data(Process* c_p,
- Port *prt,
- Eterm *retvalp)
-{
- ErtsProc2PortSigData *sigdp;
- ErtsTryImmDrvCallResult try_call_res;
- ErtsTryImmDrvCallState try_call_state
- = ERTS_INIT_TRY_IMM_DRV_CALL_STATE(
- c_p,
- prt,
- ERTS_PORT_SFLGS_INVALID_LOOKUP,
- 0,
- 0,
- am_get_data);
-
- try_call_res = try_imm_drv_call(&try_call_state);
- switch (try_call_res) {
- case ERTS_TRY_IMM_DRV_CALL_OK: {
- Eterm *hp;
- Eterm data;
- ErlHeapFragment *bp;
- Uint sz;
- if (is_immed(prt->data)) {
- bp = NULL;
- data = prt->data;
- }
- else {
- bp = new_message_buffer(prt->bp->used_size);
- data = copy_struct(prt->data,
- prt->bp->used_size,
- &hp,
- &bp->off_heap);
- }
- finalize_imm_drv_call(&try_call_state);
- if (is_immed(data))
- sz = 0;
- else
- sz = bp->used_size;
-
- hp = HAlloc(c_p, sz + 3);
- if (is_not_immed(data)) {
- data = copy_struct(data, bp->used_size, &hp, &MSO(c_p));
- free_message_buffer(bp);
- }
- *retvalp = TUPLE2(hp, am_ok, data);
- BUMP_REDS(c_p, ERTS_PORT_REDS_GET_DATA);
- return ERTS_PORT_OP_DONE;
- }
- case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT:
- return ERTS_PORT_OP_DROPPED;
- case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS:
- case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK:
- /* Schedule call instead... */
- break;
- }
-
- sigdp = erts_port_task_alloc_p2p_sig_data();
- sigdp->flags = ERTS_P2P_SIG_TYPE_GET_DATA;
-
- return erts_schedule_proc2port_signal(c_p,
- prt,
- c_p->common.id,
- retvalp,
- sigdp,
- 0,
- port_sig_get_data);
-}
-
typedef struct {
int to;
void *arg;
diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c
index 2451f41a82..301ce2d0e2 100644
--- a/erts/emulator/drivers/common/inet_drv.c
+++ b/erts/emulator/drivers/common/inet_drv.c
@@ -678,8 +678,8 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n)
#define INET_LOPT_UDP_READ_PACKETS 33 /* Number of packets to read */
#define INET_OPT_RAW 34 /* Raw socket options */
#define INET_LOPT_TCP_SEND_TIMEOUT_CLOSE 35 /* auto-close on send timeout or not */
-#define INET_LOPT_TCP_MSGQ_HIWTRMRK 36 /* set local high watermark */
-#define INET_LOPT_TCP_MSGQ_LOWTRMRK 37 /* set local low watermark */
+#define INET_LOPT_MSGQ_HIWTRMRK 36 /* set local msgq high watermark */
+#define INET_LOPT_MSGQ_LOWTRMRK 37 /* set local msgq low watermark */
/* SCTP options: a separate range, from 100: */
#define SCTP_OPT_RTOINFO 100
#define SCTP_OPT_ASSOCINFO 101
@@ -5476,27 +5476,25 @@ static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
}
continue;
- case INET_LOPT_TCP_MSGQ_HIWTRMRK:
- if (desc->stype == SOCK_STREAM) {
- ErlDrvSizeT high;
- if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
- || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
- return -1;
- high = (ErlDrvSizeT) ival;
- erl_drv_busy_msgq_limits(desc->port, NULL, &high);
- }
+ case INET_LOPT_MSGQ_HIWTRMRK: {
+ ErlDrvSizeT high;
+ if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+ || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+ return -1;
+ high = (ErlDrvSizeT) ival;
+ erl_drv_busy_msgq_limits(desc->port, NULL, &high);
continue;
+ }
- case INET_LOPT_TCP_MSGQ_LOWTRMRK:
- if (desc->stype == SOCK_STREAM) {
- ErlDrvSizeT low;
- if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
- || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
- return -1;
- low = (ErlDrvSizeT) ival;
- erl_drv_busy_msgq_limits(desc->port, &low, NULL);
- }
+ case INET_LOPT_MSGQ_LOWTRMRK: {
+ ErlDrvSizeT low;
+ if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+ || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+ return -1;
+ low = (ErlDrvSizeT) ival;
+ erl_drv_busy_msgq_limits(desc->port, &low, NULL);
continue;
+ }
case INET_LOPT_TCP_SEND_TIMEOUT:
if (desc->stype == SOCK_STREAM) {
@@ -6398,31 +6396,23 @@ static ErlDrvSSizeT inet_fill_opts(inet_descriptor* desc,
}
continue;
- case INET_LOPT_TCP_MSGQ_HIWTRMRK:
- if (desc->stype == SOCK_STREAM) {
- ErlDrvSizeT high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
- *ptr++ = opt;
- erl_drv_busy_msgq_limits(desc->port, NULL, &high);
- ival = high > INT_MAX ? INT_MAX : (int) high;
- put_int32(ival, ptr);
- }
- else {
- TRUNCATE_TO(0,ptr);
- }
+ case INET_LOPT_MSGQ_HIWTRMRK: {
+ ErlDrvSizeT high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+ *ptr++ = opt;
+ erl_drv_busy_msgq_limits(desc->port, NULL, &high);
+ ival = high > INT_MAX ? INT_MAX : (int) high;
+ put_int32(ival, ptr);
continue;
+ }
- case INET_LOPT_TCP_MSGQ_LOWTRMRK:
- if (desc->stype == SOCK_STREAM) {
- ErlDrvSizeT low = ERL_DRV_BUSY_MSGQ_READ_ONLY;
- *ptr++ = opt;
- erl_drv_busy_msgq_limits(desc->port, &low, NULL);
- ival = low > INT_MAX ? INT_MAX : (int) low;
- put_int32(ival, ptr);
- }
- else {
- TRUNCATE_TO(0,ptr);
- }
+ case INET_LOPT_MSGQ_LOWTRMRK: {
+ ErlDrvSizeT low = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+ *ptr++ = opt;
+ erl_drv_busy_msgq_limits(desc->port, &low, NULL);
+ ival = low > INT_MAX ? INT_MAX : (int) low;
+ put_int32(ival, ptr);
continue;
+ }
case INET_LOPT_TCP_SEND_TIMEOUT:
if (desc->stype == SOCK_STREAM) {
@@ -7471,6 +7461,20 @@ static void inet_stop(inet_descriptor* desc)
FREE(desc);
}
+static void set_default_msgq_limits(ErlDrvPort port)
+{
+ ErlDrvSizeT q_high = INET_HIGH_MSGQ_WATERMARK;
+ ErlDrvSizeT q_low = INET_LOW_MSGQ_WATERMARK;
+ if (q_low < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+ q_low = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+ else if (q_low > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+ q_low = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+ if (q_high < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+ q_high = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+ else if (q_high > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+ q_high = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+ erl_drv_busy_msgq_limits(port, &q_low, &q_high);
+}
/* Allocate descriptor */
static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol)
@@ -8091,9 +8095,8 @@ static int tcp_inet_init(void)
/* initialize the TCP descriptor */
-static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
+static ErlDrvData prep_tcp_inet_start(ErlDrvPort port, char* args)
{
- ErlDrvSizeT q_low, q_high;
tcp_descriptor* desc;
DEBUGF(("tcp_inet_start(%ld) {\r\n", (long)port));
@@ -8103,17 +8106,6 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
return ERL_DRV_ERROR_ERRNO;
desc->high = INET_HIGH_WATERMARK;
desc->low = INET_LOW_WATERMARK;
- q_high = INET_HIGH_MSGQ_WATERMARK;
- q_low = INET_LOW_MSGQ_WATERMARK;
- if (q_low < ERL_DRV_BUSY_MSGQ_LIM_MIN)
- q_low = ERL_DRV_BUSY_MSGQ_LIM_MIN;
- else if (q_low > ERL_DRV_BUSY_MSGQ_LIM_MAX)
- q_low = ERL_DRV_BUSY_MSGQ_LIM_MAX;
- if (q_high < ERL_DRV_BUSY_MSGQ_LIM_MIN)
- q_high = ERL_DRV_BUSY_MSGQ_LIM_MIN;
- else if (q_high > ERL_DRV_BUSY_MSGQ_LIM_MAX)
- q_high = ERL_DRV_BUSY_MSGQ_LIM_MAX;
- erl_drv_busy_msgq_limits(port, &q_low, &q_high);
desc->send_timeout = INET_INFINITY;
desc->send_timeout_close = 0;
desc->busy_on_send = 0;
@@ -8130,6 +8122,12 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
return (ErlDrvData) desc;
}
+static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
+{
+ ErlDrvData data = prep_tcp_inet_start(port, args);
+ set_default_msgq_limits(port);
+ return data;
+}
/* Copy a descriptor, by creating a new port with same settings
* as the descriptor desc.
* return NULL on error (SYSTEM_LIMIT no ports avail)
@@ -8141,7 +8139,7 @@ static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s,
ErlDrvPort port = desc->inet.port;
tcp_descriptor* copy_desc;
- copy_desc = (tcp_descriptor*) tcp_inet_start(port, NULL);
+ copy_desc = (tcp_descriptor*) prep_tcp_inet_start(port, NULL);
/* Setup event if needed */
if ((copy_desc->inet.s = s) != INVALID_SOCKET) {
@@ -9864,12 +9862,15 @@ static int should_use_so_bsdcompat(void)
* as the descriptor desc.
* return NULL on error (ENFILE no ports avail)
*/
+static ErlDrvData packet_inet_start(ErlDrvPort port, char* args, int protocol);
+
static udp_descriptor* sctp_inet_copy(udp_descriptor* desc, SOCKET s, int* err)
{
+ ErlDrvSizeT q_low, q_high;
ErlDrvPort port = desc->inet.port;
udp_descriptor* copy_desc;
- copy_desc = (udp_descriptor*) sctp_inet_start(port, NULL);
+ copy_desc = (udp_descriptor*) packet_inet_start(port, NULL, IPPROTO_SCTP);
/* Setup event if needed */
if ((copy_desc->inet.s = s) != INVALID_SOCKET) {
@@ -9900,9 +9901,17 @@ static udp_descriptor* sctp_inet_copy(udp_descriptor* desc, SOCKET s, int* err)
FREE(copy_desc);
return NULL;
}
+
+ /* Read busy msgq limits of parent */
+ q_low = q_high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+ erl_drv_busy_msgq_limits(desc->inet.port, &q_low, &q_high);
+ /* Write same busy msgq limits to child */
+ erl_drv_busy_msgq_limits(port, &q_low, &q_high);
+
copy_desc->inet.port = port;
copy_desc->inet.dport = driver_mk_port(port);
*err = 0;
+
return copy_desc;
}
#endif
@@ -9935,13 +9944,17 @@ static ErlDrvData packet_inet_start(ErlDrvPort port, char* args, int protocol)
static ErlDrvData udp_inet_start(ErlDrvPort port, char *args)
{
- return packet_inet_start(port, args, IPPROTO_UDP);
+ ErlDrvData data = packet_inet_start(port, args, IPPROTO_UDP);
+ set_default_msgq_limits(port);
+ return data;
}
#ifdef HAVE_SCTP
static ErlDrvData sctp_inet_start(ErlDrvPort port, char *args)
{
- return packet_inet_start(port, args, IPPROTO_SCTP);
+ ErlDrvData data = packet_inet_start(port, args, IPPROTO_SCTP);
+ set_default_msgq_limits(port);
+ return data;
}
#endif
diff --git a/erts/emulator/sys/win32/sys.c b/erts/emulator/sys/win32/sys.c
index e0422de026..922967c698 100755
--- a/erts/emulator/sys/win32/sys.c
+++ b/erts/emulator/sys/win32/sys.c
@@ -399,11 +399,12 @@ int* pBuild; /* Pointer to build number. */
* Definitions for driver flags.
*/
-#define DF_OVR_READY 1 /* Overlapped result is ready. */
-#define DF_EXIT_THREAD 2 /* The thread should exit. */
-#define DF_XLAT_CR 4 /* The thread should translate CRs. */
-#define DF_DROP_IF_INVH 8 /* Drop packages instead of crash if
+#define DF_OVR_READY 1 /* Overlapped result is ready. */
+#define DF_EXIT_THREAD 2 /* The thread should exit. */
+#define DF_XLAT_CR 4 /* The thread should translate CRs. */
+#define DF_DROP_IF_INVH 8 /* Drop packages instead of crash if
invalid handle (stderr) */
+#define DF_THREAD_FLUSHED 16 /* The thread should exit. */
#define OV_BUFFER_PTR(dp) ((LPVOID) ((dp)->ov.Internal))
#define OV_NUM_TO_READ(dp) ((dp)->ov.InternalHigh)
@@ -2141,8 +2142,9 @@ threaded_writer(LPVOID param)
for (;;) {
handle = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
- if (aio->flags & DF_EXIT_THREAD)
+ if (aio->flags & DF_EXIT_THREAD) {
break;
+ }
buf = OV_BUFFER_PTR(aio);
numToWrite = OV_NUM_TO_READ(aio);
@@ -2150,6 +2152,7 @@ threaded_writer(LPVOID param)
if (handle == (WAIT_OBJECT_0 + 1) && numToWrite == 0) {
SetEvent(aio->flushReplyEvent);
+ aio->flags |= DF_THREAD_FLUSHED;
continue;
}
@@ -2197,6 +2200,7 @@ threaded_writer(LPVOID param)
if (aio->flags & DF_EXIT_THREAD)
break;
}
+ aio->flags |= DF_THREAD_FLUSHED;
CloseHandle(aio->fd);
aio->fd = INVALID_HANDLE_VALUE;
unrefer_driver_data(aio->dp);
@@ -2340,9 +2344,11 @@ static void fd_stop(ErlDrvData data)
(void) driver_select(dp->port_num,
(ErlDrvEvent)dp->out.ov.hEvent,
ERL_DRV_WRITE, 0);
- ASSERT(dp->out.flushEvent);
- SetEvent(dp->out.flushEvent);
- WaitForSingleObject(dp->out.flushReplyEvent, INFINITE);
+ do {
+ ASSERT(dp->out.flushEvent);
+ SetEvent(dp->out.flushEvent);
+ } while (WaitForSingleObject(dp->out.flushReplyEvent, 10) == WAIT_TIMEOUT
+ || !(dp->out.flags & DF_THREAD_FLUSHED));
}
}
@@ -2433,12 +2439,12 @@ threaded_exiter(LPVOID param)
*/
i = 0;
if (dp->out.thread != (HANDLE) -1) {
- dp->out.flags = DF_EXIT_THREAD;
+ dp->out.flags |= DF_EXIT_THREAD;
SetEvent(dp->out.ioAllowed);
handles[i++] = dp->out.thread;
}
if (dp->in.thread != (HANDLE) -1) {
- dp->in.flags = DF_EXIT_THREAD;
+ dp->in.flags |= DF_EXIT_THREAD;
SetEvent(dp->in.ioAllowed);
handles[i++] = dp->in.thread;
}
diff --git a/erts/emulator/test/busy_port_SUITE.erl b/erts/emulator/test/busy_port_SUITE.erl
index 2c63296b83..4b4af0babe 100644
--- a/erts/emulator/test/busy_port_SUITE.erl
+++ b/erts/emulator/test/busy_port_SUITE.erl
@@ -170,6 +170,7 @@ send_3(Config) when is_list(Config) ->
?line {Owner,Slave} = get_slave(),
?line ok = erlang:send(Slave, {Owner,{command,"set busy"}},
[nosuspend]),
+ receive after 100 -> ok end, % ensure command reached port
?line nosuspend = erlang:send(Slave, {Owner,{command,"busy"}},
[nosuspend]),
?line unlock_slave(),
@@ -563,6 +564,7 @@ scheduling_delay_busy_nosuspend(Config) ->
{2,{call,[{var,1},open_port]}},
{0,{cast,[{var,1},{command,1,100}]}},
{0,{cast,[{var,1},{busy,2}]}},
+ {0,{timer,sleep,[200]}}, % ensure reached port
{10,{call,[{var,1},{command,3,[nosuspend]}]}},
{0,{timer,sleep,[200]}},
{0,{erlang,port_command,[{var,2},<<$N>>,[force]]}},
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-16 22:48:46 +0000