erts: Add port monitors

* erlang:monitor/2 with port argument is added, erlang:demonitor, using port task API and avoiding locking;
* port_info and process_info support for monitored ports (with named port monitors support);
* Exit signals contain type 'process' or 'port';
* Propagation of port exit signals;
* Self-cleaning when origin process dies with monitor on;
* 8 test cases + testcase for port driver crashing;
* Documentation for all of the above (monitor, demonitor, port_info and process_info) updated

1 files changed, 269 insertions, 141 deletions

diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index b18910e2c7..fc14061a44 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -282,20 +282,17 @@ res_no_proc: {
     }
 }
 
-#define ERTS_DEMONITOR_FALSE		2
-#define ERTS_DEMONITOR_TRUE		1
-#define ERTS_DEMONITOR_BADARG		0
-#define ERTS_DEMONITOR_YIELD_TRUE	-1
-#define ERTS_DEMONITOR_INTERNAL_ERROR	-2
-
-static int
+/* This function is allowed to return range of values handled by demonitor/1-2
+ * Namely: atoms true, false, yield, internal_error, badarg or THE_NON_VALUE
+ */
+static Eterm
 remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to)
 {
     ErtsDSigData dsd;
     ErtsMonitor *dmon;
     ErtsMonitor *mon;
     int code;
-    int res;
+    Eterm res = am_false;
 #ifndef ERTS_SMP
     int stale_mon = 0;
 #endif
@@ -328,7 +325,7 @@ remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to)
 	mon = erts_remove_monitor(&ERTS_P_MONITORS(c_p), ref);
 	erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_LINK);
 
-	res = ERTS_DEMONITOR_TRUE;
+        res = am_true;
 	break;
 
     case ERTS_DSIG_PREP_CONNECTED:
@@ -352,7 +349,7 @@ remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to)
 	     * This is possible when smp support is enabled.
 	     * 'DOWN' message just arrived.
 	     */
-	    res = ERTS_DEMONITOR_TRUE;
+            res = am_true;
 	}
 	else {
 	    /*
@@ -367,16 +364,13 @@ remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to)
 					     : mon->pid), 
 					    ref,
 					    0);
-	    res = (code == ERTS_DSIG_SEND_YIELD
-		   ? ERTS_DEMONITOR_YIELD_TRUE
-		   : ERTS_DEMONITOR_TRUE);
+            res = (code == ERTS_DSIG_SEND_YIELD ? am_yield : am_true);
 	    erts_destroy_monitor(dmon);
-
 	}
 	break;
     default:
 	ASSERT(! "Invalid dsig prepare result");
-	return ERTS_DEMONITOR_INTERNAL_ERROR;
+        return am_internal_error;
     }
 
 #ifndef ERTS_SMP
@@ -404,27 +398,96 @@ remote_demonitor(Process *c_p, DistEntry *dep, Eterm ref, Eterm to)
     return res;
 }
 
-static int demonitor(Process *c_p, Eterm ref, Eterm *multip)
+static ERTS_INLINE void
+demonitor_local_process(Process *c_p, Eterm ref, Eterm to, Eterm *res)
+{
+    Process *rp = erts_pid2proc_opt(c_p,
+                           ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_LINK,
+                           to,
+                           ERTS_PROC_LOCK_LINK,
+                           ERTS_P2P_FLG_ALLOW_OTHER_X);
+    ErtsMonitor *mon = erts_remove_monitor(&ERTS_P_MONITORS(c_p), ref);
+
+#ifndef ERTS_SMP
+    ASSERT(mon);
+#else
+    if (!mon)
+        *res = am_false;
+    else
+#endif
+    {
+        *res = am_true;
+        erts_destroy_monitor(mon);
+    }
+    if (rp) {
+        ErtsMonitor *rmon;
+        rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), ref);
+        if (rp != c_p)
+            erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+        if (rmon != NULL)
+            erts_destroy_monitor(rmon);
+    }
+    else {
+        ERTS_SMP_ASSERT_IS_NOT_EXITING(c_p);
+    }
+}
+
+static ERTS_INLINE BIF_RETTYPE
+demonitor_local_port(Process *origin, Eterm ref, Eterm target)
 {
-   ErtsMonitor *mon = NULL;  /* The monitor entry to delete */
-   Process  *rp;    /* Local target process */
-   Eterm     to = NIL;    /* Monitor link traget */
-   DistEntry *dep = NULL;  /* Target's distribution entry */
-   int deref_de = 0;
-   int res;
-   int unlock_link = 1;
+    BIF_RETTYPE res = am_false;
+    Port        *port = erts_port_lookup_raw(target);
+
+    if (!port) {
+        BIF_ERROR(origin, BADARG);
+    }
+    erts_smp_proc_unlock(origin, ERTS_PROC_LOCK_LINK);
+
+    if (port) {
+        Eterm trap_ref;
+        switch (erts_port_demonitor(origin, ERTS_PORT_DEMONITOR_NORMAL,
+                                    port, ref, &trap_ref)) {
+        case ERTS_PORT_OP_DROPPED:
+        case ERTS_PORT_OP_BADARG:
+            break;
+        case ERTS_PORT_OP_SCHEDULED:
+            BIF_TRAP3(await_port_send_result_trap, origin, trap_ref,
+                      am_busy_port, am_true);
+            /* the busy_port atom will never be returned, because it cannot be
+             * returned from erts_port_(de)monitor, but just in case if in future
+             * internal API changes - you may see this atom */
+        default:
+            break;
+        }
+    }
+    else {
+        ERTS_SMP_ASSERT_IS_NOT_EXITING(origin);
+    }
+    BIF_RET(res);
+}
 
+/* Can return atom true, false, yield, internal_error, badarg or
+ * THE_NON_VALUE if error occured or trap has been set up
+ */
+static
+BIF_RETTYPE demonitor(Process *c_p, Eterm ref, Eterm *multip)
+{
+   ErtsMonitor  *mon = NULL;  /* The monitor entry to delete */
+   Eterm        to = NIL;     /* Monitor link traget */
+   DistEntry    *dep = NULL;  /* Target's distribution entry */
+   int          deref_de = 0;
+   BIF_RETTYPE  res = am_false;
+   int          unlock_link = 1;
 
    erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_LINK);
 
    if (is_not_internal_ref(ref)) {
-       res = ERTS_DEMONITOR_BADARG;
+       res = am_badarg;
        goto done; /* Cannot be this monitor's ref */
    }
 
    mon = erts_lookup_monitor(ERTS_P_MONITORS(c_p), ref);
    if (!mon) {
-       res = ERTS_DEMONITOR_FALSE;
        goto done;
    }
 
@@ -432,70 +495,50 @@ static int demonitor(Process *c_p, Eterm ref, Eterm *multip)
    case MON_TIME_OFFSET:
        *multip = am_true;
        erts_demonitor_time_offset(ref);
-       res = ERTS_DEMONITOR_TRUE;
+       res = am_true;
        break;
    case MON_ORIGIN:
        to = mon->pid;
        *multip = am_false;
        if (is_atom(to)) {
-	   /* Monitoring a name at node to */
-	   ASSERT(is_node_name_atom(to));
-	   dep = erts_sysname_to_connected_dist_entry(to);
-	   ASSERT(dep != erts_this_dist_entry);
-	   if (dep)
-	       deref_de = 1;
+           /* Monitoring a name at node to */
+           ASSERT(is_node_name_atom(to));
+           dep = erts_sysname_to_connected_dist_entry(to);
+           ASSERT(dep != erts_this_dist_entry);
+           if (dep)
+               deref_de = 1;
+       } else if (is_port(to)) {
+           if (port_dist_entry(to) != erts_this_dist_entry) {
+               goto badarg;
+           }
+           res = demonitor_local_port(c_p, ref, to);
+           unlock_link = 0;
+           goto done;
        } else {
-	   ASSERT(is_pid(to));
-	   dep = pid_dist_entry(to);
+           ASSERT(is_pid(to));
+           dep = pid_dist_entry(to);
        }
        if (dep != erts_this_dist_entry) {
-	   res = remote_demonitor(c_p, dep, ref, to);
-	   /* remote_demonitor() unlocks link lock on c_p */
-	   unlock_link = 0;
+           res = remote_demonitor(c_p, dep, ref, to);
+           /* remote_demonitor() unlocks link lock on c_p */
+           unlock_link = 0;
        }
        else { /* Local monitor */
-	   if (deref_de) {
-	       deref_de = 0;
-	       erts_deref_dist_entry(dep);
-	   }
-	   dep = NULL;
-	   rp = erts_pid2proc_opt(c_p,
-				  ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_LINK,
-				  to,
-				  ERTS_PROC_LOCK_LINK,
-				  ERTS_P2P_FLG_ALLOW_OTHER_X);
-	   mon = erts_remove_monitor(&ERTS_P_MONITORS(c_p), ref);
-#ifndef ERTS_SMP
-	   ASSERT(mon);
-#else
-	   if (!mon)
-	       res = ERTS_DEMONITOR_FALSE;
-	   else
-#endif
-	   {
-	       res = ERTS_DEMONITOR_TRUE;
-	       erts_destroy_monitor(mon);
-	   }
-	   if (rp) {
-	       ErtsMonitor *rmon;
-	       rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), ref);
-	       if (rp != c_p)
-		   erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
-	       if (rmon != NULL)
-		   erts_destroy_monitor(rmon);
-	   }
-	   else {
-	       ERTS_SMP_ASSERT_IS_NOT_EXITING(c_p);
-	   }
-
+           if (deref_de) {
+               deref_de = 0;
+               erts_deref_dist_entry(dep);
+           }
+           dep = NULL;
+           demonitor_local_process(c_p, ref, to, &res);
        }
        break;
-   default:
-       res = ERTS_DEMONITOR_BADARG;
+    default /* case */ :
+badarg:
+       res = am_badarg; /* will be converted to error by caller */
        *multip = am_false;
        break;
    }
- done:
+done:
 
    if (unlock_link)
        erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_LINK);
@@ -506,21 +549,20 @@ static int demonitor(Process *c_p, Eterm ref, Eterm *multip)
    }
 
    ERTS_SMP_LC_ASSERT(ERTS_PROC_LOCK_MAIN == erts_proc_lc_my_proc_locks(c_p));
-   return res;
+   BIF_RET(res);
 }
 
 BIF_RETTYPE demonitor_1(BIF_ALIST_1)
 {
     Eterm multi;
     switch (demonitor(BIF_P, BIF_ARG_1, &multi)) {
-    case ERTS_DEMONITOR_FALSE:
-    case ERTS_DEMONITOR_TRUE:
-	BIF_RET(am_true);
-    case ERTS_DEMONITOR_YIELD_TRUE:
-	ERTS_BIF_YIELD_RETURN(BIF_P, am_true);
-    case ERTS_DEMONITOR_BADARG:
-	BIF_ERROR(BIF_P, BADARG);
-    case ERTS_DEMONITOR_INTERNAL_ERROR:
+    case am_false:
+    case am_true:       BIF_RET(am_true);
+    case THE_NON_VALUE: BIF_RET(THE_NON_VALUE);
+    case am_yield:      ERTS_BIF_YIELD_RETURN(BIF_P, am_true);
+    case am_badarg:     BIF_ERROR(BIF_P, BADARG);
+
+    case am_internal_error:
     default:
 	ASSERT(! "demonitor(): internal error");
 	BIF_ERROR(BIF_P, EXC_INTERNAL_ERROR);
@@ -529,11 +571,11 @@ BIF_RETTYPE demonitor_1(BIF_ALIST_1)
 
 BIF_RETTYPE demonitor_2(BIF_ALIST_2)
 {
-    Eterm res = am_true;
-    Eterm multi = am_false;
-    int info = 0;
-    int flush = 0;
-    Eterm list = BIF_ARG_2;
+    BIF_RETTYPE res = am_true;
+    Eterm       multi = am_false;
+    int         info = 0;
+    int         flush = 0;
+    Eterm       list = BIF_ARG_2;
 
     while (is_list(list)) {
 	Eterm* consp = list_val(list);
@@ -554,24 +596,27 @@ BIF_RETTYPE demonitor_2(BIF_ALIST_2)
 	goto badarg;
 
     switch (demonitor(BIF_P, BIF_ARG_1, &multi)) {
-    case ERTS_DEMONITOR_FALSE:
+    case THE_NON_VALUE:
+        /* If other error occured or trap has been set up - pass through */
+        BIF_RET(THE_NON_VALUE);
+    case am_false:
 	if (info)
 	    res = am_false;
 	if (flush) {
-	flush_messages:
+flush_messages:
 	    BIF_TRAP3(flush_monitor_messages_trap, BIF_P,
 		      BIF_ARG_1, multi, res);
 	}
-    case ERTS_DEMONITOR_TRUE:
+    case am_true:
 	if (multi == am_true && flush)
 	    goto flush_messages;
 	BIF_RET(res);
-    case ERTS_DEMONITOR_YIELD_TRUE:
+    case am_yield:
 	ERTS_BIF_YIELD_RETURN(BIF_P, am_true);
-    case ERTS_DEMONITOR_BADARG:
-    badarg:
+    case am_badarg:
+badarg:
 	BIF_ERROR(BIF_P, BADARG);
-    case ERTS_DEMONITOR_INTERNAL_ERROR:
+    case am_internal_error:
     default:
 	ASSERT(! "demonitor(): internal error");
 	BIF_ERROR(BIF_P, EXC_INTERNAL_ERROR);
@@ -615,14 +660,13 @@ erts_queue_monitor_message(Process *p,
     erts_queue_message(p, *p_locksp, msgp, tup, am_system);
 }
 
-static BIF_RETTYPE
+static Eterm
 local_pid_monitor(Process *p, Eterm target, Eterm mon_ref, int boolean)
 {
-    BIF_RETTYPE ret;
-    Process *rp;
+    Eterm         ret = mon_ref;
+    Process      *rp;
     ErtsProcLocks p_locks = ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_LINK;
 
-    ERTS_BIF_PREP_RET(ret, mon_ref);
     if (target == p->common.id) {
 	return ret;
     }
@@ -658,40 +702,112 @@ local_pid_monitor(Process *p, Eterm target, Eterm mon_ref, int boolean)
 }
 
 static BIF_RETTYPE
-local_name_monitor(Process *p, Eterm target_name)
+local_port_monitor(Process *origin, Eterm target)
 {
-    BIF_RETTYPE ret;
-    Eterm mon_ref;
-    ErtsProcLocks p_locks = ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_LINK;
-    Process *rp;
+    BIF_RETTYPE ref = erts_make_ref(origin);
+    Port *port = erts_sig_lookup_port(origin, target);
+    ErtsProcLocks p_locks = ERTS_PROC_LOCK_MAIN;
 
-    mon_ref = erts_make_ref(p);
-    ERTS_BIF_PREP_RET(ret, mon_ref);
-    erts_smp_proc_lock(p, ERTS_PROC_LOCK_LINK);
-    rp = erts_whereis_process(p, p_locks, target_name, ERTS_PROC_LOCK_LINK,
-			      ERTS_P2P_FLG_ALLOW_OTHER_X);
-    if (!rp) {
-	DeclareTmpHeap(lhp,3,p);
+    if (!port) {
+res_no_proc:
+        /* Send the DOWN message immediately. Ref is made on the fly because
+         * caller has never seen it yet. */
+        erts_queue_monitor_message(origin, &p_locks, ref,
+                                   am_port, target, am_noproc);
+    }
+    else {
+        switch (erts_port_monitor(origin, port, target, &ref)) {
+        case ERTS_PORT_OP_DROPPED:
+        case ERTS_PORT_OP_BADARG:
+            goto res_no_proc;
+        case ERTS_PORT_OP_SCHEDULED:
+            BIF_TRAP3(await_port_send_result_trap, origin, ref,
+                      am_busy_port, ref);
+            /* the busy_port atom will never be returned, because it cannot be
+             * returned from erts_port_monitor, but just in case if in future
+             * internal API changes - you may see this atom */
+        default:
+            break;
+        }
+    }
+    erts_smp_proc_unlock(origin, p_locks & ~ERTS_PROC_LOCK_MAIN);
+    BIF_RET(ref);
+}
+
+/* Type = process | port :: atom(), 1st argument passed to erlang:monitor/2
+ */
+static BIF_RETTYPE
+local_name_monitor(Process *self, Eterm type, Eterm target_name)
+{
+    BIF_RETTYPE ret = erts_make_ref(self);
+
+    ErtsProcLocks p_locks = ERTS_PROC_LOCK_MAIN | ERTS_PROC_LOCK_LINK;
+    Process     *proc = NULL;
+    Port        *port = NULL;
+
+    erts_smp_proc_lock(self, ERTS_PROC_LOCK_LINK);
+
+    erts_whereis_name(self, p_locks, target_name,
+                      &proc, ERTS_PROC_LOCK_LINK,
+                      ERTS_P2P_FLG_ALLOW_OTHER_X,
+                      &port, 0);
+
+    /* If the name is not registered,
+     * or if we asked for proc and got a port,
+     * or if we asked for port and got a proc,
+     * we just send the 'DOWN' message.
+     */
+    if ((!proc && !port) ||
+        (type == am_process && port) ||
+        (type == am_port && proc)) {
+        DeclareTmpHeap(lhp,3,self);
 	Eterm item;
-	UseTmpHeap(3,p);
-	erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
+        UseTmpHeap(3,self);
+
+        erts_smp_proc_unlock(self, ERTS_PROC_LOCK_LINK);
 	p_locks &= ~ERTS_PROC_LOCK_LINK;
+
 	item = TUPLE2(lhp, target_name, erts_this_dist_entry->sysname);
-	erts_queue_monitor_message(p, &p_locks,
-				   mon_ref, am_process, item, am_noproc);
-	UnUseTmpHeap(3,p);
-    }
-    else if (rp != p) {
-	erts_add_monitor(&ERTS_P_MONITORS(p), MON_ORIGIN, mon_ref, rp->common.id,
-			 target_name);
-	erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, mon_ref, p->common.id,
-			 target_name);
-	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
+        erts_queue_monitor_message(self, &p_locks,
+                                   ret,
+                                   type, /* = process|port :: atom() */
+                                   item, am_noproc);
+        UnUseTmpHeap(3,self);
+    }
+    else if (port) {
+        erts_smp_proc_unlock(self, p_locks & ~ERTS_PROC_LOCK_MAIN);
+        p_locks &= ~ERTS_PROC_LOCK_MAIN;
+
+        switch (erts_port_monitor(self, port, target_name, &ret)) {
+        case ERTS_PORT_OP_DONE:
+            return ret;
+        case ERTS_PORT_OP_SCHEDULED: { /* Scheduled a signal */
+            ASSERT(is_internal_ref(ret));
+            BIF_TRAP3(await_port_send_result_trap, self,
+                      ret, am_true, ret);
+            /* bif_trap returns */
+        } break;
+        default:
+            goto badarg;
+        }
+    }
+    else if (proc != self) {
+        erts_add_monitor(&ERTS_P_MONITORS(self), MON_ORIGIN, ret,
+                         proc->common.id, target_name);
+        erts_add_monitor(&ERTS_P_MONITORS(proc), MON_TARGET, ret,
+                         self->common.id, target_name);
+        erts_smp_proc_unlock(proc, ERTS_PROC_LOCK_LINK);
     }
 
-    erts_smp_proc_unlock(p, p_locks & ~ERTS_PROC_LOCK_MAIN);
-
-    return ret;
+    if (p_locks) {
+        erts_smp_proc_unlock(self, p_locks & ~ERTS_PROC_LOCK_MAIN);
+    }
+    BIF_RET(ret);
+badarg:
+    if (p_locks) {
+        erts_smp_proc_unlock(self, p_locks & ~ERTS_PROC_LOCK_MAIN);
+    }
+    BIF_ERROR(self, BADARG);
 }
 
 static BIF_RETTYPE
@@ -758,7 +874,7 @@ remote_monitor(Process *p, Eterm bifarg1, Eterm bifarg2,
 	break;
     }
 
-    return ret;
+    BIF_RET(ret);
 }
 	
 BIF_RETTYPE monitor_2(BIF_ALIST_2)
@@ -772,8 +888,9 @@ BIF_RETTYPE monitor_2(BIF_ALIST_2)
     switch (BIF_ARG_1) {
     case am_time_offset: {
 	Eterm ref;
-	if (BIF_ARG_2 != am_clock_service)
-	    goto error;
+        if (BIF_ARG_2 != am_clock_service) {
+            goto badarg;
+        }
 	ref = erts_make_ref(BIF_P);
 	erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_LINK);
 	erts_add_monitor(&ERTS_P_MONITORS(BIF_P), MON_TIME_OFFSET,
@@ -783,46 +900,57 @@ BIF_RETTYPE monitor_2(BIF_ALIST_2)
 	BIF_RET(ref);
     }
     case am_process:
+    case am_port:
 	break;
     default:
-	goto error;
+        goto badarg;
     }
 
-    if (is_internal_pid(target)) {
-    local_pid:
-	ret = local_pid_monitor(BIF_P, target, erts_make_ref(BIF_P), 0);
-    } else if (is_external_pid(target)) {
+    if (is_internal_pid(target) && BIF_ARG_1 == am_process) {
+local_pid:
+        ret = local_pid_monitor(BIF_P, target, erts_make_ref(BIF_P), 0);
+    } else if (is_external_pid(target) && BIF_ARG_1 == am_process) {
 	dep = external_pid_dist_entry(target);
 	if (dep == erts_this_dist_entry)
 	    goto local_pid;
 	ret = remote_monitor(BIF_P, BIF_ARG_1, BIF_ARG_2, dep, target, 0);
+    } else if (is_internal_port(target) && BIF_ARG_1 == am_port) {
+local_port:
+        ret = local_port_monitor(BIF_P, target);
+    } else if (is_external_port(target) && BIF_ARG_1 == am_port) {
+        dep = external_port_dist_entry(target);
+        if (dep == erts_this_dist_entry) {
+            goto local_port;
+        }
+        goto badarg; /* No want remote port */
     } else if (is_atom(target)) {
-	ret = local_name_monitor(BIF_P, target);
+        ret = local_name_monitor(BIF_P, BIF_ARG_1, target);
     } else if (is_tuple(target)) {
 	Eterm *tp = tuple_val(target);
 	Eterm remote_node;
 	Eterm name;
-	if (arityval(*tp) != 2) 
-	    goto error;
+        if (arityval(*tp) != 2) {
+            goto badarg;
+        }
 	remote_node = tp[2];
 	name = tp[1];
 	if (!is_atom(remote_node) || !is_atom(name)) {
-	    goto error;
+            goto badarg;
 	}
 	if (!erts_is_alive && remote_node != am_Noname) {
-	    goto error; /* Remote monitor from (this) undistributed node */
+            goto badarg; /* Remote monitor from (this) undistributed node */
 	}
 	dep = erts_sysname_to_connected_dist_entry(remote_node);
 	if (dep == erts_this_dist_entry) {
 	    deref_de = 1;
-	    ret = local_name_monitor(BIF_P, name);
+            ret = local_name_monitor(BIF_P, BIF_ARG_1, name);
 	} else {
 	    if (dep)
 		deref_de = 1;
 	    ret = remote_monitor(BIF_P, BIF_ARG_1, BIF_ARG_2, dep, name, 1);
 	}
     } else {
-    error:
+badarg:
 	ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
     }
     if (deref_de) {