erts: Add enif_port_command

6 files changed, 266 insertions, 16 deletions

diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index 21bf752bb6..6ed89780b4 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -375,6 +375,29 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
     return 1;
 }
 
+int
+enif_port_command(ErlNifEnv *env, const ErlNifPort* to_port,
+                  ErlNifEnv *msg_env, ERL_NIF_TERM msg)
+{
+
+    ErtsSchedulerData *esdp = erts_get_scheduler_data();
+    int scheduler = esdp ? esdp->no : 0;
+    Port *prt;
+
+    if (scheduler == 0 || !env)
+        return 0;
+
+    prt = erts_port_lookup(to_port->port_id,
+                           (erts_port_synchronous_ops
+                            ? ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP
+                            : ERTS_PORT_SFLGS_INVALID_LOOKUP));
+
+    if (!prt)
+        return 0;
+
+    return erts_port_output_async(prt, env->proc->common.id, msg);
+}
+
 ERL_NIF_TERM enif_make_copy(ErlNifEnv* dst_env, ERL_NIF_TERM src_term)
 {
     Uint sz;
diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h
index 0333e95331..32afb53bfc 100644
--- a/erts/emulator/beam/erl_nif_api_funcs.h
+++ b/erts/emulator/beam/erl_nif_api_funcs.h
@@ -171,6 +171,7 @@ ERL_NIF_API_FUNC_DECL(int, enif_is_port_alive, (ErlNifEnv *env, ErlNifPort *port
 ERL_NIF_API_FUNC_DECL(int, enif_get_local_port, (ErlNifEnv* env, ERL_NIF_TERM, ErlNifPort* port_id));
 ERL_NIF_API_FUNC_DECL(int, enif_term_to_binary, (ErlNifEnv *env, ERL_NIF_TERM term, ErlNifBinary *bin));
 ERL_NIF_API_FUNC_DECL(int, enif_binary_to_term, (ErlNifEnv *env, ErlNifBinary *bin, ERL_NIF_TERM *term));
+ERL_NIF_API_FUNC_DECL(int, enif_port_command, (ErlNifEnv *env, const ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg));
 
 /*
 ** ADD NEW ENTRIES HERE (before this comment) !!!
@@ -334,6 +335,7 @@ ERL_NIF_API_FUNC_DECL(int,enif_is_on_dirty_scheduler,(ErlNifEnv*));
 #  define enif_get_local_port ERL_NIF_API_FUNC_MACRO(enif_get_local_port)
 #  define enif_term_to_binary ERL_NIF_API_FUNC_MACRO(enif_term_to_binary)
 #  define enif_binary_to_term ERL_NIF_API_FUNC_MACRO(enif_binary_to_term)
+#  define enif_port_command ERL_NIF_API_FUNC_MACRO(enif_port_command)
 
 /*
 ** ADD NEW ENTRIES HERE (before this comment)
diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index fa97707a87..dbc6ead216 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -949,6 +949,8 @@ 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 *);
 
+int erts_port_output_async(Port *, Eterm, Eterm);
+
 /*
  * Signals from ports to ports. Used by sys drivers.
  */
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index e85395e062..b200344af5 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -180,10 +180,9 @@ p2p_sig_data_to_task(ErtsProc2PortSigData *sigdp)
     return ptp;
 }
 
-ErtsProc2PortSigData *
-erts_port_task_alloc_p2p_sig_data(void)
+static ERTS_INLINE ErtsProc2PortSigData *
+p2p_sig_data_init(ErtsPortTask *ptp)
 {
-    ErtsPortTask *ptp = port_task_alloc();
 
     ptp->type = ERTS_PORT_TASK_PROC_SIG;
     ptp->u.alive.flags = ERTS_PT_FLG_SIG_DEP;
@@ -194,6 +193,31 @@ erts_port_task_alloc_p2p_sig_data(void)
     return &ptp->u.alive.td.psig.data;
 }
 
+ErtsProc2PortSigData *
+erts_port_task_alloc_p2p_sig_data(void)
+{
+    ErtsPortTask *ptp = port_task_alloc();
+
+    return p2p_sig_data_init(ptp);
+}
+
+ErtsProc2PortSigData *
+erts_port_task_alloc_p2p_sig_data_extra(size_t extra, void **extra_ptr)
+{
+    ErtsPortTask *ptp = erts_alloc(ERTS_ALC_T_PORT_TASK,
+                                   sizeof(ErtsPortTask) + extra);
+
+    *extra_ptr = ptp+1;
+
+    return p2p_sig_data_init(ptp);
+}
+
+void
+erts_port_task_free_p2p_sig_data(ErtsProc2PortSigData *sigdp)
+{
+    schedule_port_task_free(p2p_sig_data_to_task(sigdp));
+}
+
 static ERTS_INLINE Eterm
 task_caller(ErtsPortTask *ptp)
 {
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index 335f7a77d5..56cef4e352 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -269,6 +269,8 @@ int erts_port_task_schedule(Eterm,
 void erts_port_task_free_port(Port *);
 int erts_port_is_scheduled(Port *);
 ErtsProc2PortSigData *erts_port_task_alloc_p2p_sig_data(void);
+ErtsProc2PortSigData *erts_port_task_alloc_p2p_sig_data_extra(size_t extra, void **extra_ptr);
+void erts_port_task_free_p2p_sig_data(ErtsProc2PortSigData *sigdp);
 
 #ifdef ERTS_SMP
 void erts_enqueue_port(ErtsRunQueue *rq, Port *pp);
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 797e4cdadc..093334aab7 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -1749,7 +1749,6 @@ cleanup_scheduled_outputv(ErlIOVec *ev, ErlDrvBinary *cbinp)
 	    driver_free_binary(ev->binv[i]);
     if (cbinp)
 	driver_free_binary(cbinp);
-    erts_free(ERTS_ALC_T_DRV_CMD_DATA, ev);
 }
 
 static int
@@ -1887,6 +1886,188 @@ port_sig_output(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *si
     return ERTS_PORT_REDS_CMD_OUTPUT;
 }
 
+
+/*
+ * This erts_port_output will always create a port task.
+ * The call is treated as a port_command call, i.e. no
+ * badsig i generated if the input in invalid. However
+ * an error_logger message is generated.
+ */
+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;
+    char *buf = NULL;
+    ErtsPortTaskHandle *ns_pthp;
+
+    if (drv->outputv) {
+        ErlIOVec ev;
+	SysIOVec* ivp;
+	ErlDrvBinary**  bvp;
+        int vsize;
+	Uint csize;
+	Uint pvsize;
+	Uint pcsize;
+        size_t iov_offset, binv_offset, alloc_size;
+        Uint blimit = 0;
+        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;
+
+            /* 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);
+            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 *);
+
+        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);
+
+        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;
+            }
+        }
+#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 */
+
+        /* Need to increase refc on all binaries */
+        for (i = 1; i < evp->vsize; i++)
+            if (bvp[i])
+                driver_binary_inc_refc(bvp[i]);
+
+	sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV;
+	sigdp->u.outputv.from = from;
+	sigdp->u.outputv.evp = evp;
+	sigdp->u.outputv.cbinp = cbin;
+	port_sig_callback = port_sig_outputv;
+    } else {
+        ErlDrvSizeT ERTS_DECLARE_DUMMY(r);
+
+	/*
+	 * Apperently there exist code that write 1 byte to
+	 * much in buffer. Where it resides I don't know, but
+	 * we can live with one byte extra allocated...
+	 */
+
+        if (erts_iolist_size(list, &size))
+            goto bad_value;
+
+        buf = erts_alloc(ERTS_ALC_T_DRV_CMD_DATA, size + 1);
+
+        r = erts_iolist_to_buf(list, buf, size);
+        ASSERT(ERTS_IOLIST_TO_BUF_SUCCEEDED(r));
+
+	sigdp = erts_port_task_alloc_p2p_sig_data();
+	sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUT;
+	sigdp->u.output.from = from;
+	sigdp->u.output.bufp = buf;
+	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);
+
+    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:
+
+    /*
+     * We call badsig directly here as this function is called with
+     * the main lock of the calling process still held.
+     * At the moment this operation is always not a bang_op, so
+     * only an error_logger message should be generated, no badsig.
+    */
+
+    badsig_received(0, prt, erts_atomic32_read_nob(&prt->state), 1);
+
+    return 0;
+
+}
+
 ErtsPortOpResult
 erts_port_output(Process *c_p,
 		 int flags,
@@ -1896,7 +2077,7 @@ erts_port_output(Process *c_p,
 		 Eterm *refp)
 {
     ErtsPortOpResult res;
-    ErtsProc2PortSigData *sigdp;
+    ErtsProc2PortSigData *sigdp = NULL;
     erts_driver_t *drv = prt->drv_ptr;
     size_t size;
     int try_call;
@@ -1978,10 +2159,13 @@ erts_port_output(Process *c_p,
 	    evp = &ev;
 	}
 	else {
-	    char *ptr = erts_alloc((try_call
-				    ? ERTS_ALC_T_TMP
-				    : ERTS_ALC_T_DRV_CMD_DATA), alloc_size);
-
+            char *ptr;
+            if (try_call) {
+                ptr = erts_alloc(ERTS_ALC_T_TMP, alloc_size);
+            } else {
+                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);
@@ -2010,9 +2194,12 @@ erts_port_output(Process *c_p,
 	bvp[0] = NULL;
 	evp->vsize = io_list_to_vec(list, ivp+1, bvp+1, cbin, blimit);
 	if (evp->vsize < 0) {
-	    if (evp != &ev)
-		erts_free(try_call ? ERTS_ALC_T_TMP : ERTS_ALC_T_DRV_CMD_DATA,
-			  evp);
+            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);
 	    goto bad_value;
 	}
@@ -2064,8 +2251,10 @@ erts_port_output(Process *c_p,
 		/* Fall through... */
 	    case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT:
 		driver_free_binary(cbin);
-		if (evp != &ev)
+		if (evp != &ev) {
+                    ASSERT(!sigdp);
 		    erts_free(ERTS_ALC_T_TMP, evp);
+                }
 		if (try_call_res != ERTS_TRY_IMM_DRV_CALL_OK)
 		    return ERTS_PORT_OP_DROPPED;
 		if (c_p)
@@ -2076,8 +2265,10 @@ erts_port_output(Process *c_p,
 		if (async_nosuspend
 		    && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) {
 		    driver_free_binary(cbin);
-		    if (evp != &ev)
+                    if (evp != &ev) {
+                        ASSERT(!sigdp);
 			erts_free(ERTS_ALC_T_TMP, evp);
+                    }
 		    return ((sched_flags & ERTS_PTS_FLG_EXIT)
 			    ? ERTS_PORT_OP_DROPPED
 			    : ERTS_PORT_OP_BUSY);
@@ -2092,9 +2283,16 @@ erts_port_output(Process *c_p,
 		if (bvp[i])
 		    driver_binary_inc_refc(bvp[i]);
 
-	    new_evp = erts_alloc(ERTS_ALC_T_DRV_CMD_DATA, alloc_size);
+            /* The port task and iovec is allocated in the
+               same structure as an optimization. This
+               is especially important in erts_port_output_async
+               of when !try_call */
+            ASSERT(sigdp == NULL);
+            sigdp = erts_port_task_alloc_p2p_sig_data_extra(
+                alloc_size, (void**)&new_evp);
 
 	    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);
@@ -2142,7 +2340,6 @@ erts_port_output(Process *c_p,
 	    evp = new_evp;
 	}
 
-	sigdp = erts_port_task_alloc_p2p_sig_data();
 	sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV;
 	sigdp->u.outputv.from = from;
 	sigdp->u.outputv.evp = evp;