Implement true asynchronous signaling between processes and ports

1 files changed, 53 insertions, 3 deletions

diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index 71b1cc6e25..c41f8104c7 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -27,6 +27,9 @@
 #define ERTS_PORT_TASK_H_BASIC_TYPES__
 #include "erl_sys_driver.h"
 #include "erl_smp.h"
+#define ERL_PORT_GET_PORT_TYPE_ONLY__
+#include "erl_port.h"
+#undef ERL_PORT_GET_PORT_TYPE_ONLY__
 typedef erts_smp_atomic_t ErtsPortTaskHandle;
 #endif
 
@@ -43,12 +46,19 @@ typedef erts_smp_atomic_t ErtsPortTaskHandle;
 #define ERTS_INCLUDE_SCHEDULER_INTERNALS
 #endif
 
+#define ERTS_PT_FLG_WAIT_BUSY		(1 << 0)
+#define ERTS_PT_FLG_SIG_DEP		(1 << 1)
+#define ERTS_PT_FLG_NOSUSPEND		(1 << 2)
+#define ERTS_PT_FLG_REF			(1 << 3)
+#define ERTS_PT_FLG_BAD_OUTPUT		(1 << 4)
+
 typedef enum {
     ERTS_PORT_TASK_INPUT,
     ERTS_PORT_TASK_OUTPUT,
     ERTS_PORT_TASK_EVENT,
     ERTS_PORT_TASK_TIMEOUT,
-    ERTS_PORT_TASK_DIST_CMD
+    ERTS_PORT_TASK_DIST_CMD,
+    ERTS_PORT_TASK_PROC_SIG
 } ErtsPortTaskType;
 
 #ifdef ERTS_INCLUDE_SCHEDULER_INTERNALS
@@ -61,14 +71,35 @@ extern erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
 #define ERTS_PTS_FLG_HAVE_TASKS		(((erts_aint32_t) 1) << 2)
 #define ERTS_PTS_FLG_EXIT		(((erts_aint32_t) 1) << 3)
 #define ERTS_PTS_FLG_BUSY		(((erts_aint32_t) 1) << 4)
+#define ERTS_PTS_FLG_HAVE_BUSY_TASKS	(((erts_aint32_t) 1) << 5)
+#define ERTS_PTS_FLG_HAVE_NS_TASKS	(((erts_aint32_t) 1) << 6)
+#define ERTS_PTS_FLG_PARALLELISM	(((erts_aint32_t) 1) << 7)
+#define ERTS_PTS_FLG_FORCE_SCHED	(((erts_aint32_t) 1) << 8)
+
 
+#define ERTS_PTS_FLGS_FORCE_SCHEDULE_OP		\
+    (ERTS_PTS_FLG_EXIT				\
+     | ERTS_PTS_FLG_HAVE_BUSY_TASKS		\
+     | ERTS_PTS_FLG_HAVE_TASKS			\
+     | ERTS_PTS_FLG_EXEC			\
+     | ERTS_PTS_FLG_FORCE_SCHED)
 
 typedef struct ErtsPortTask_ ErtsPortTask;
+typedef struct ErtsPortTaskBusyCallerTable_ ErtsPortTaskBusyCallerTable;
+typedef struct ErtsPortTaskHandleList_ ErtsPortTaskHandleList;
 
 typedef struct {
     Port *next;
     struct {
-	ErtsPortTask *local;
+	struct {
+	    struct {
+		ErtsPortTask *first;
+		ErtsPortTask *last;
+		ErtsPortTaskBusyCallerTable *table;
+		ErtsPortTaskHandleList *nosuspend;
+	    } busy;
+	    ErtsPortTask *first;
+	} local;
 	struct {
 	    ErtsPortTask *first;
 	    ErtsPortTask *last;
@@ -87,6 +118,7 @@ ERTS_GLB_INLINE void erts_port_task_init_sched(ErtsPortTaskSched *ptsp,
 ERTS_GLB_INLINE void erts_port_task_fini_sched(ErtsPortTaskSched *ptsp);
 ERTS_GLB_INLINE void erts_port_task_sched_lock(ErtsPortTaskSched *ptsp);
 ERTS_GLB_INLINE void erts_port_task_sched_unlock(ErtsPortTaskSched *ptsp);
+ERTS_GLB_INLINE int erts_port_task_sched_lock_is_locked(ErtsPortTaskSched *ptsp);
 
 #ifdef ERTS_INCLUDE_SCHEDULER_INTERNALS
 ERTS_GLB_INLINE int erts_port_task_have_outstanding_io_tasks(void);
@@ -110,7 +142,11 @@ ERTS_GLB_INLINE void
 erts_port_task_init_sched(ErtsPortTaskSched *ptsp, Eterm instr_id)
 {
     ptsp->next = NULL;
-    ptsp->taskq.local = NULL;
+    ptsp->taskq.local.busy.first = NULL;
+    ptsp->taskq.local.busy.last = NULL;
+    ptsp->taskq.local.busy.table = NULL;
+    ptsp->taskq.local.busy.nosuspend = NULL;
+    ptsp->taskq.local.first = NULL;
     ptsp->taskq.in.first = NULL;
     ptsp->taskq.in.last = NULL;
     erts_smp_atomic32_init_nob(&ptsp->flags, 0);
@@ -135,6 +171,17 @@ erts_port_task_sched_unlock(ErtsPortTaskSched *ptsp)
 #endif
 }
 
+ERTS_GLB_INLINE int
+erts_port_task_sched_lock_is_locked(ErtsPortTaskSched *ptsp)
+{
+#if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)
+    return erts_lc_mtx_is_locked(&ptsp->mtx);
+#else
+    return 0;
+#endif
+}
+
+
 ERTS_GLB_INLINE void
 erts_port_task_fini_sched(ErtsPortTaskSched *ptsp)
 {
@@ -162,12 +209,15 @@ void erts_port_task_init(void);
 #endif
 
 int erts_port_task_abort(ErtsPortTaskHandle *);
+void erts_port_task_abort_nosuspend_tasks(Port *);
+
 int erts_port_task_schedule(Eterm,
 			    ErtsPortTaskHandle *,
 			    ErtsPortTaskType,
 			    ...);
 void erts_port_task_free_port(Port *);
 int erts_port_is_scheduled(Port *);
+ErtsProc2PortSigData *erts_port_task_alloc_p2p_sig_data(void);
 
 #ifdef ERTS_SMP
 void erts_enqueue_port(ErtsRunQueue *rq, Port *pp);