Do not treat port_set_data/port_get_data as signals

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;