Add driver callback epilogue

4 files changed, 76 insertions, 67 deletions

diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index beeddd09a0..1eb6dea710 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -271,6 +271,30 @@ extern erts_smp_atomic_t erts_bytes_in;		/* no bytes sent into the system */
   (ERTS_PORT_SFLGS_INVALID_LOOKUP					\
    | ERTS_PORT_SFLG_DISTRIBUTION)
 
+
+/*
+ * Costs in reductions for some port operations.
+ */
+#define ERTS_PORT_REDS_EXECUTE		10
+#define ERTS_PORT_REDS_FREE		100
+#define ERTS_PORT_REDS_TIMEOUT		400
+#define ERTS_PORT_REDS_INPUT		400
+#define ERTS_PORT_REDS_OUTPUT		400
+#define ERTS_PORT_REDS_EVENT		400
+#define ERTS_PORT_REDS_CMD_OUTPUTV	400
+#define ERTS_PORT_REDS_CMD_OUTPUT	400
+#define ERTS_PORT_REDS_EXIT		300
+#define ERTS_PORT_REDS_CONNECT		40
+#define ERTS_PORT_REDS_UNLINK		40
+#define ERTS_PORT_REDS_LINK		40
+#define ERTS_PORT_REDS_BADSIG		40
+#define ERTS_PORT_REDS_CONTROL		400
+#define ERTS_PORT_REDS_CALL		400
+#define ERTS_PORT_REDS_INFO		100
+#define ERTS_PORT_REDS_SET_DATA		40
+#define ERTS_PORT_REDS_GET_DATA		40
+#define ERTS_PORT_REDS_TERMINATE	200
+
 void print_port_info(Port *, int, void *);
 void erts_port_free(Port *);
 #ifndef ERTS_SMP
@@ -278,7 +302,7 @@ void erts_port_cleanup(Port *);
 #endif
 void erts_fire_port_monitor(Port *prt, Eterm ref);
 #ifdef ERTS_SMP
-void erts_smp_xports_unlock(Port *);
+int erts_port_handle_xports(Port *);
 #endif
 
 #if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)
@@ -387,6 +411,9 @@ erts_smp_port_unlock(Port *prt)
 extern const Port erts_invalid_port;
 #define ERTS_PORT_LOCK_BUSY ((Port *) &erts_invalid_port)
 
+int erts_is_port_ioq_empty(Port *);
+void erts_terminate_port(Port *);
+
 #ifdef ERTS_SMP
 Port *erts_de2port(DistEntry *, Process *, ErtsProcLocks);
 #endif
@@ -409,6 +436,7 @@ ERTS_GLB_INLINE Eterm erts_drvport2id(ErlDrvPort);
 ERTS_GLB_INLINE Uint32 erts_portid2status(Eterm);
 ERTS_GLB_INLINE int erts_is_port_alive(Eterm);
 ERTS_GLB_INLINE int erts_is_valid_tracer_port(Eterm);
+ERTS_GLB_INLINE int erts_port_driver_callback_epilogue(Port *, erts_aint32_t *);
 
 #if ERTS_GLB_INLINE_INCL_FUNC_DEF
 
@@ -700,6 +728,37 @@ erts_is_valid_tracer_port(Eterm id)
 {
     return !(erts_portid2status(id) & ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP);
 }
+
+ERTS_GLB_INLINE int
+erts_port_driver_callback_epilogue(Port *prt, erts_aint32_t *statep)
+{
+    int reds = 0;
+    erts_aint32_t state;
+
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+    state = erts_atomic32_read_nob(&prt->state);
+    if ((state & ERTS_PORT_SFLG_CLOSING) && erts_is_port_ioq_empty(prt)) {
+	reds += ERTS_PORT_REDS_TERMINATE;
+	erts_terminate_port(prt);
+	state = erts_atomic32_read_nob(&prt->state);
+	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+    }
+
+#ifdef ERTS_SMP
+    if (prt->xports) {
+	reds += erts_port_handle_xports(prt);
+	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+	ASSERT(!prt->xports);
+    }
+#endif
+
+    if (statep)
+	*statep = state;
+
+    return reds;
+}
+
 #endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */
 
 struct binary;
@@ -774,19 +833,6 @@ typedef int (*ErtsProc2PortSigCallback)(Port *,
 					int,
 					ErtsProc2PortSigData *);
 
-#define ERTS_PORT_REDS_CMD_OUTPUTV	400
-#define ERTS_PORT_REDS_CMD_OUTPUT	400
-#define ERTS_PORT_REDS_EXIT		300
-#define ERTS_PORT_REDS_CONNECT		40
-#define ERTS_PORT_REDS_UNLINK		40
-#define ERTS_PORT_REDS_LINK		40
-#define ERTS_PORT_REDS_BADSIG		40
-#define ERTS_PORT_REDS_CONTROL		400
-#define ERTS_PORT_REDS_CALL		400
-#define ERTS_PORT_REDS_INFO		100
-#define ERTS_PORT_REDS_SET_DATA		40
-#define ERTS_PORT_REDS_GET_DATA		40
-
 typedef enum {
     ERTS_PORT_OP_BADARG,
     ERTS_PORT_OP_CALLER_EXIT,
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index e8ed00daee..b6b113b753 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -49,18 +49,6 @@ static void chk_task_queues(Port *pp, ErtsPortTask *execq, int processing_busy_q
 #define ERTS_PT_DBG_CHK_TASK_QS(PP, EQ, PBQ)
 #endif
 
-
-/*
- * Costs in reductions for some port operations.
- */
-#define ERTS_PORT_REDS_EXECUTE		10
-#define ERTS_PORT_REDS_FREE		100
-#define ERTS_PORT_REDS_TIMEOUT		400
-#define ERTS_PORT_REDS_INPUT		400
-#define ERTS_PORT_REDS_OUTPUT		400
-#define ERTS_PORT_REDS_EVENT		400
-#define ERTS_PORT_REDS_TERMINATE	200
-
 #ifdef USE_VM_PROBES
 #define DTRACE_DRIVER(PROBE_NAME, PP)                              \
     if (DTRACE_ENABLED(PROBE_NAME)) {                              \
@@ -1473,22 +1461,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 	    break;
 	}
 
-	state = erts_atomic32_read_nob(&pp->state);
-	if ((state & ERTS_PORT_SFLG_CLOSING) && erts_is_port_ioq_empty(pp)) {
-	    reds += ERTS_PORT_REDS_TERMINATE;
-	    erts_terminate_port(pp);
-	    state = erts_atomic32_read_nob(&pp->state);
-	}
-
-	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp));
-
-#ifdef ERTS_SMP
-	if (pp->xports)
-	    erts_smp_xports_unlock(pp);
-	ASSERT(!pp->xports);
-#endif
-
-	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp));
+	reds += erts_port_driver_callback_epilogue(pp, &state);
 
     aborted_port_task:
 	schedule_port_task_free(ptp);
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index d91dd7d9ec..06b90b38ab 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -913,8 +913,6 @@ int erts_add_driver_entry(ErlDrvEntry *drv, DE_Handle *handle, int driver_list_l
 void erts_destroy_driver(erts_driver_t *drv);
 int erts_save_suspend_process_on_port(Port*, Process*);
 Port *erts_open_driver(erts_driver_t*, Eterm, char*, SysDriverOpts*, int *, int *);
-int erts_is_port_ioq_empty(Port *);
-void erts_terminate_port(Port *);
 void erts_init_io(int, int);
 void erts_raw_port_command(Port*, byte*, Uint);
 void driver_report_exit(ErlDrvPort, int);
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 2a55a7c09f..9024e9ab52 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -680,7 +680,7 @@ erts_open_driver(erts_driver_t* driver,	/* Pointer to driver. */
 	}
 #ifdef ERTS_SMP
 	if (port->xports)
-	    erts_smp_xports_unlock(port);
+	    erts_port_handle_xports(port);
 	ASSERT(!port->xports);
 #endif
     }
@@ -826,9 +826,9 @@ driver_create_port(ErlDrvPort creator_port_ix, /* Creating port */
 }
 
 #ifdef ERTS_SMP
-void
-erts_smp_xports_unlock(Port *prt)
+int erts_port_handle_xports(Port *prt)
 {
+    int reds = 0;
     ErtsXPortsList *xplp;
 
     ASSERT(prt);
@@ -839,16 +839,20 @@ erts_smp_xports_unlock(Port *prt)
 	ErtsXPortsList *free_xplp;
 	erts_aint32_t state;
 	if (rprt->xports)
-	    erts_smp_xports_unlock(rprt);
+	    reds += erts_port_handle_xports(rprt);
 	state = erts_atomic32_read_nob(&rprt->state);
-	if ((state & ERTS_PORT_SFLG_CLOSING) && erts_is_port_ioq_empty(rprt))
+	if ((state & ERTS_PORT_SFLG_CLOSING) && erts_is_port_ioq_empty(rprt)) {
 	    terminate_port(rprt);
+	    reds += ERTS_PORT_REDS_TERMINATE;
+	}
 	erts_port_release(rprt);
 	free_xplp = xplp;
 	xplp = xplp->next;
 	xports_list_free(free_xplp);
+	reds++;
     }
     prt->xports = NULL;
+    return reds;
 }
 #endif
 
@@ -1234,13 +1238,9 @@ finalize_imm_drv_call(ErtsTryImmDrvCallState *sp)
     Port *prt = sp->port;
     Process *c_p = sp->c_p;
 
-    erts_unblock_fpe(sp->fpe_was_unmasked);
+    erts_port_driver_callback_epilogue(prt, NULL);
 
-#ifdef ERTS_SMP
-    if (prt->xports)
-	erts_smp_xports_unlock(prt);
-    ASSERT(!prt->xports);
-#endif
+    erts_unblock_fpe(sp->fpe_was_unmasked);
 
     if (IS_TRACED_FL(prt, F_TRACE_SCHED_PORTS))
 	trace_sched_ports_where(prt, am_out, sp->port_op);
@@ -3110,7 +3110,7 @@ static void flush_port(Port *p)
 	}
 #ifdef ERTS_SMP
 	if (p->xports)
-	    erts_smp_xports_unlock(p);
+	    erts_port_handle_xports(p);
 	ASSERT(!p->xports);
 #endif
     }
@@ -3165,7 +3165,7 @@ terminate_port(Port *prt)
 	erts_unblock_fpe(fpe_was_unmasked);
 #ifdef ERTS_SMP
 	if (prt->xports)
-	    erts_smp_xports_unlock(prt);
+	    erts_port_handle_xports(prt);
 	ASSERT(!prt->xports);
 #endif
     }
@@ -4757,9 +4757,7 @@ int async_ready(Port *p, void* data)
     ERTS_SMP_CHK_NO_PROC_LOCKS;
 
     if (p) {
-	erts_aint32_t state = erts_atomic32_read_nob(&p->state);
 	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p));
-	ASSERT(!(state & ERTS_PORT_SFLGS_DEAD));
 	if (p->drv_ptr->ready_async != NULL) {
 #ifdef USE_VM_PROBES
             if (DTRACE_ENABLED(driver_ready_async)) {
@@ -4769,15 +4767,9 @@ int async_ready(Port *p, void* data)
 #endif
 	    (*p->drv_ptr->ready_async)((ErlDrvData)p->drv_data, data);
 	    need_free = 0;
-#ifdef ERTS_SMP
-	    if (p->xports)
-		erts_smp_xports_unlock(p);
-	    ASSERT(!p->xports);
-#endif
-	}
-	if ((state & ERTS_PORT_SFLG_CLOSING) && is_port_ioq_empty(p)) {
-	    terminate_port(p);
+
 	}
+	erts_port_driver_callback_epilogue(p, NULL);
     }
     return need_free;
 }