aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam')
-rw-r--r--erts/emulator/beam/bif.tab6
-rw-r--r--erts/emulator/beam/erl_alloc.types1
-rwxr-xr-xerts/emulator/beam/erl_bif_info.c4
-rw-r--r--erts/emulator/beam/erl_bif_port.c173
-rw-r--r--erts/emulator/beam/erl_node_tables.c6
-rw-r--r--erts/emulator/beam/erl_port.h19
-rw-r--r--erts/emulator/beam/io.c229
7 files changed, 160 insertions, 278 deletions
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/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..9edbd2776d 100755
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -2861,12 +2861,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);
- 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_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..4095b88369 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 \
@@ -856,10 +857,6 @@ struct ErtsProc2PortSigData_ {
struct {
Eterm from;
} unlink;
- struct {
- ErlHeapFragment *bp;
- Eterm data;
- } set_data;
} u;
} ;
@@ -953,7 +950,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/io.c b/erts/emulator/beam/io.c
index 43918a7141..d32e7b5d21 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;
@@ -3371,11 +3370,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);
@@ -4502,225 +4498,6 @@ erts_port_info(Process* c_p,
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;